Storm读取Mysql数据库表写入Mysql数据库另一张表

1、spout：

package com.TestStorm;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.Reader;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Map;

import org.apache.storm.generated.DistributedRPCInvocations.Processor.result;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

public class TestSpout extends BaseRichSpout {

	String url = "jdbc:mysql://192.168.0.44:3306/mysql";
	String username = "root";
	String password = "mysql";

	private ResultSet rs;
	private Statement state;
	private SpoutOutputCollector collector;
	int id = 0;
	public void nextTuple() {
		try {
			
			String str = "";	
			if (rs.next()) {
				id = rs.getInt(1);
				str = rs.getString(1) + " " + rs.getString(2) + "\n";
				System.out.print(str);
				collector.emit(new Values(str));
			}else{

				rs.close();
				rs = state.executeQuery("select * from storm  where id > "+ id +" order by id limit 10000" );
			
			}
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

	}

	public void open(Map arg0, TopologyContext arg1, SpoutOutputCollector arg2) {

		try {
			String driver = "com.mysql.jdbc.Driver";
			Class.forName(driver);
			Connection conn = DriverManager.getConnection("jdbc:mysql://192.168.0.44:3306/mysql", "root", "mysql");
			state = conn.createStatement();
			rs = state.executeQuery("select * from storm where id > "+ id +" order by id limit 10000");

			this.collector = arg2;
		} catch (ClassNotFoundException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public void declareOutputFields(OutputFieldsDeclarer arg0) {

		arg0.declare(new Fields("str"));
	}
}package com.TestStorm;

import ja