Flume自定义Source

 1 public class MQSource extends AbstractSource implements EventDrivenSource, Configurable {
 2 
 3     private Logger logger = org.slf4j.LoggerFactory.getLogger(MQSource.class);
 4 
 5     private long heartbeat;
 6 
 7     private MQReceiver receiver;
 8 
 9     private HandleLineCallBack handle;
10 
11     private Thread t;
12 
13     @Override
14     public void configure(Context context) {
15 
16         String mq_url = context.getString(MQContext.MQ_BROKER_URI, ActiveMQConnection.DEFAULT_BROKER_URL);
17         String mq_userName = context.getString(MQContext.MQ_USERNAME, ActiveMQConnection.DEFAULT_USER);
18         String mq_password = context.getString(MQContext.MQ_PASSWORD, ActiveMQConnection.DEFAULT_PASSWORD);
19         String mq_queueKey = context.getString(MQContext.MQ_QUEUEKEY, "NULL");
20         String handleClass = context.getString(MQContext.HANDLECLASS, "NULL");
21 
22         long mq_reciveTimeout = context.getLong(MQContext.MQ_RECIVETIMEOUT, 3000L);
23         long heartbeat = context.getLong(MQContext.HEARTBEAT, 3000L);
24         this.heartbeat = heartbeat;
25         if ("NULL".equals(mq_queueKey)) {
26             logger.error("{} :  Unable to load MQ_queueKey ", getName());
27             return;
28         }
29         if ("NULL".equals(handleClass)) {
30             logger.warn("{} :  Unable to handleClass using DefaultHandle ", getName());
31             handleClass = "com.bidlink.handle.DefaultHandle";
32         }
33 
34         MQConfig mqconfig = new MQConfig(mq_url, mq_userName, mq_password, mq_queueKey, mq_reciveTimeout);
35         logger.info("{}  MQ Configuration : {} ", getName(), mqconfig.toString());
36         receiver = MQFactory.MQ.getReceiver(mqconfig);
37         logger.info("{} .get recerver key is {} .  obj is  : {} ", getName(), mqconfig.getQueueKey(), receiver);
38 
39         try {
40             @SuppressWarnings("unchecked")
41             Class<?> handleClazz = (Class<? extends HandleLineCallBack>) Class.forName(handleClass);
42             handle = (HandleLineCallBack) handleClazz.newInstance();
43         } catch (ClassNotFoundException e) {
44             logger.error("{} unable to load class  {}  . {} ", getName(), handleClass, e);
45         } catch (InstantiationException e1) {
46             logger.error("{} instance class error {} . {} ", getName(), handleClass, e1);
47         } catch (IllegalAccessException e2) {
48             logger.error("{} occur  exception {} . {} ", getName(), handleClass, e2);
49         }
50     }
51 
52     @Override
53     public synchronized void start() {
54         logger.info("MQSource start.....");
55         // TODO Auto-generated method stub
56         try {
57             t = new Thread() {
58                 public void run() {
59                     while (true) {
60                         try {
61                             List<String> lines = receiver.getText();
62                             for (String line : lines) {
63                                 //logger.info("Message line : {} ",line);
64                                 Event e = new SimpleEvent();
65                                 String refStr = handle.refactor(line);
66                                 e.setBody(refStr.getBytes("GBK"));
67                                 getChannelProcessor().processEvent(e);
68                             }
69                             super.start();
70                             Thread.sleep(heartbeat);
71                         } catch (Exception e1) {
72                             e1.printStackTrace();
73                         }
74                         
75                     }
76                 };
77             };
78             t.start();
79         } catch (Exception e1) {
80             logger.error("error starting MQResource {} ",e1.getMessage());
81             e1.printStackTrace();
82         }
83     }
84 
85     @Override
86     public synchronized void stop() {
87         logger.info("MQSource stoping...");
88         if (t.isAlive()) {
89             try {
90                 t.join();
91             } catch (InterruptedException e) {
92                 e.printStackTrace();
93             }
94             t.interrupt();
95         }
96         super.stop();
97     }
98 
99 }

  Event e = new SimpleEvent();
                              
  e.setBody("hello everyone ".getBytes("GBK"));
 
  getChannelProcessor().processEvent(e);

  super.start();

tier1.sources.testSources.type = org.yunjume.source.MQSource
tier1.sources.testSources.MQ_userName= admin
tier1.sources.testSources.MQ_password= admin123
tier1.sources.testSources.MQ_brokerURL=tcp://localhost:61616
tier1.sources.testSources.MQ_queueKey=FirstQueue
tier1.sources.testSources.MQ_reciveTimeout=30000
tier1.sources.testSources.heartbeat=30000
# to process mq message queue line and return new line . 
tier1.sources.testSources.handleClass=org.yunjume.handle.DefaultHandle
tier1.sources.testSources.channels = testChannels