Kafka多线程消费消息创建线程池并封装成Jar包

XML及属性配置

1.业务处理(kiafka.worker.xml)

1<?xml version="1.0" encoding="UTF-8"?>
2<root>
3<topics>
4    <!--name:要消费的topic;-->
5    <!--worker:执行消费逻辑的worker,配置值为执行的类全路径-->
6    <!--consumerThreads:队列获取线程的数量,建议和broker数量一致;-->
7    <!--workerThreads:队列执行逻辑线程的数量,建议按消费队列量修改-->
8    <topic topic="TEST_TOPIC" worker="com.zzj.worker.ServiceWorker" consumerThreads="3" workerThreads="20"/>
9</topics>
10</root>

2.消费者连接属性kafka.consumer.properties

1zookeeper.connect=your broker1:2181,your broker2:2181,your broker3:2181
2group.id=sddzj
3zookeeper.session.timeout.ms=1000
4zookeeper.sync.time.ms=500
5auto.commit.interval.ms=1000

Jar包结构

1.自定义消息处理父类

1public abstract class Worker<T> {
2
3    private String topic;//订阅主题public abstract void execute(T message);//业务处理接口
4
5    public String getTopic() {
6        return topic;
7    }
8
9    public void setTopic(String topic) {
10        this.topic = topic;
11    }
12
13}

2.业务处理类（继承消息处理父类）

1public class ServiceWorker extends Worker<T> {
2
3   
4    @Override
5    public void execute(T message) {
6        //业务处理，即取到消息中的内容，根据消息内容做对应处理
7       
8  
9    }
10
11}

3.消费者接收消息

1public class KafkaConsumer implements Runnable {
2
3    private KafkaStream stream;
4    private int consumerThreadNumber;
5    private Worker worker;
6    private ExecutorService workerExecute;
7
8    public KafkaConsumer(KafkaStream stream, int consumerThreadNumber, Worker worker, int workerThreadNum) {
9        this.consumerThreadNumber = consumerThreadNumber;
10        this.stream = stream;
11        this.worker = worker;
12        this.workerExecute = Executors.newFixedThreadPool(workerThreadNum);
13    }
14
15    public void run() {
16        ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
17        while (it.hasNext()) {
18            String message = new String(it.next().message());
19                workerExecute.submit(new Runnable() {
20                    public void run() {
21                        try {
22                            worker.execute(message);
23                        } catch (Exception ex) {
24                            
25                        }
26                    }
27                });
28
29        }
30    }
31
32}

4.创建topic消费线程池

1private static void startWorker(HierarchicalConfiguration topicConfig, Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap) {
2        String topic = topicConfig.getString("[@topic]");
3        String workerCls = topicConfig.getString("[@worker]");
4        Integer consumerThreadNum = topicConfig.getInt("[@consumerThreads]");
5        Integer workerThreadNum = topicConfig.getInt("[@workerThreads]");
6
7        List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
8
9        //创建每个topic的消费线程池
10        ExecutorService consumerExecutor = Executors.newFixedThreadPool(consumerThreadNum);
11        //获取消费的worker
12        Worker worker = null;
13        try {
14            Class<Worker> cls = (Class<Worker>)Class.forName(workerCls);
15
16            if (cls != null){
17                worker = cls.newInstance();
18            }
19        } catch (Exception e) {
20           
21        }
22
23        if (worker == null )  return;
24
25        worker.setTopic(topic);
26
27        for (final KafkaStream stream : streams) {
28            consumerExecutor.submit(new KafkaConsumer(stream, consumerThreadNum, worker, workerThreadNum));
29        }
30    }