消费者

apache kafka工具类，消费者Consumer类

public class Consumer { private ConsumerHandler handler; private ConsumerConfig config; private KafkaConsumer<String, String> consumer; private boolean startFlag = false; /** * 创建消费者 * * @param handler * 消费者处理类 * @param config * 消费者处理配置 */ public Consumer(ConsumerHandler handler, ConsumerConfig config) { this.handler = handler; this.config = config; init(); } /** * 初始化接收器 */ private void init() { Properties props = new Properties(); props.put("bootstrap.servers", config.getBootstrapServers());// 服务器ip:端口号，集群用逗号分隔 props.put("group.id", config.getGroupID()); /* 是否自动确认offset */ props.put("enable.auto.commit", "true"); /* 自动确认offset的时间间隔 */ props.put("auto.commit.interval.ms", config.getAutoCommitInterVal()); props.put("session.timeout.ms", config.getSessionTimeOut()); /* key的序列化类 */ props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); /* value的序列化类 */ props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); consumer = new KafkaConsumer<>(props); if (config.isProcessBeforeData()) { /* 消费者订阅的topic, 可同时订阅多个 */ consumer.subscribe(config.getTopicList(), new ConsumerRebalanceListener() { @Override public void onPartitionsRevoked(Collection<TopicPartition> partitions) { } @Override public void onPartitionsAssigned(Collection<TopicPartition> partitions) { for (TopicPartition partition : partitions) { long offset = handler.getSeek(partition.topic(), partition.partition()); if (offset >= 0) { if (consumer != null) { consumer.seek(partition, offset + 1); } } else { consumer.seekToBeginning(partitions); } } } }); start(); } else { consumer.subscribe(config.getTopicList()); } } public void start() { startFlag = true; while (startFlag) { /* 读取数据，读取超时时间为XXms */ ConsumerRecords<String, String> records = consumer.poll(config.getPollTime()); if (records.count() > 0) { long offset = 0; int partition = 0; for (ConsumerRecord<String, String> record : records) { if (record != null) { offset = record.offset(); partition = record.partition(); try { handler.processObject(record.topic(), record.partition(), record.offset(), record.value()); } catch (Exception e) { e.printStackTrace(); } } } } try { Thread.currentThread(); Thread.sleep(1000); } catch (Exception e) { e.printStackTrace(); } } consumer.close(); } public void stop() { startFlag = false; } } 

消费者配置ConsumerConfig类

public class ConsumerConfig { private String bootstrapServers; private String groupID; private int autoCommitInterVal =1000; private int sessionTimeOut = 30000; private List<String> topicList; private boolean processBeforeData; private long pollTime = 100; public ConsumerConfig() { super(); } /** * 创建消费者配置 * @param bootstrapServers 服务器配合 格式为服务器ip:端口号，集群用逗号分隔 例如 192.168.1.1:9092,192.168.1.2:9092 * @param groupID groupID * @param autoCommitInterVal 自动提交时间单位毫秒， 默认1000 * @param sessionTimeOut 超时时间单位毫秒 ， 默认30000 * @param topicList topicList列表 * @param processBeforeData 是否处理启动之前的数据，该开关需要配置consumerHandler的跨步存储使用 * @param pollTime 每次获取数据等待时间单位毫秒，默认100毫秒 */ public ConsumerConfig(String bootstrapServers, String groupID, int autoCommitInterVal, int sessionTimeOut ,List<String> topicList,boolean processBeforeData,long pollTime) { this.bootstrapServers = bootstrapServers; this.groupID = groupID; this.autoCommitInterVal = autoCommitInterVal; this.sessionTimeOut = sessionTimeOut; this.topicList = topicList; this.processBeforeData = processBeforeData; this.pollTime = pollTime; } public String getBootstrapServers() { return bootstrapServers; } public void setBootstrapServers(String bootstrapServers) { this.bootstrapServers = bootstrapServers; } public String getGroupID() { return groupID; } public void setGroupID(String groupID) { this.groupID = groupID; } public int getAutoCommitInterVal() { return autoCommitInterVal; } public void setAutoCommitInterVal(int autoCommitInterVal) { this.autoCommitInterVal = autoCommitInterVal; } public int getSessionTimeOut() { return sessionTimeOut; } public void setSessionTimeOut(int sessionTimeOut) { this.sessionTimeOut = sessionTimeOut; } public List<String> getTopicList() { return topicList; } public void setTopicList(List<String> topicList) { this.topicList = topicList; } 

public boolean isProcessBeforeData() { return processBeforeData; } public void setProcessBeforeData(boolean processBeforeData) { this.processBeforeData = processBeforeData; } public long getPollTime() { return pollTime; } public void setPollTime(long pollTime) { this.pollTime = pollTime; } } 

消费者处理ConsumerHandler类

public interface ConsumerHandler { /** * 处理收到的消息 * @param topic 收到消息的topic名称 * @param partition 收到消息的partition内容 * @param offset 收到消息在队列中的编号 * @param value 收到的消息 */ void processObject(String topic,int partition,long offset,String value); /** * 获取跨步 * @param topic 接受消息的topic * @param partition 接受消息的partition * @return 当前topic,partition下的seek */ long getSeek(String topic , int partition); } 

生产者

kafka生产者，工具Producer类

public class Producer { private ProducerConfig config ; private org.apache.kafka.clients.producer.Producer<String,String> producer; public Producer(ProducerConfig config){ this.config = config; init(); } private void init(){ Properties props = new Properties(); props.put("bootstrap.servers",config.getBootstrapServers()); props.put("acks", "all"); props.put("retries", config.getRetries()); props.put("batch.size", config.getBatchSize()); props.put("linger.ms", config.getLingerMs()); props.put("buffer.memory", config.getBufferMemory()); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); producer = new KafkaProducer<>(props); } 

/** * 发送消息 * @param topic 要发送的topic * @param msg */ public void sendMessage(String topic,String msg){ try { producer.send(new ProducerRecord<String, String>(config.getTopic(), String.valueOf(new Date().getTime()), msg)).get(); } catch (InterruptedException e) { // TODO 自动生成的 catch 块 e.printStackTrace(); } catch (ExecutionException e) { // TODO 自动生成的 catch 块 e.printStackTrace(); } producer.flush(); } public void close(){ producer.close(); } } 

kafka生产者配置ProducerConfig类

public class ProducerConfig { private String bootstrapServers; private String topic; private int retries = 0; private int batchSize = 16384; private int lingerMs=1; private int bufferMemory=33554432; public ProducerConfig() { super(); } /** * 创建生产者配置文件 * @param bootstrapServers 服务器配合 格式为服务器ip:端口号，集群用逗号分隔 例如 192.168.1.1:9092,192.168.1.2:9092 * @param retries * @param batchSize * @param lingerMs * @param bufferMemory */ public ProducerConfig(String bootstrapServers,int retries, int batchSize, int lingerMs, int bufferMemory) { this.bootstrapServers = bootstrapServers; this.retries = retries; this.batchSize = batchSize; this.lingerMs = lingerMs; this.bufferMemory = bufferMemory; } public String getBootstrapServers() { return bootstrapServers; } public void setBootstrapServers(String bootstrapServers) { this.bootstrapServers = bootstrapServers; } public String getTopic() { return topic; } public void setTopic(String topic) { this.topic = topic; } public int getRetries() { return retries; } public void setRetries(int retries) { this.retries = retries; } public int getBatchSize() { return batchSize; } public void setBatchSize(int batchSize) { this.batchSize = batchSize; } public int getLingerMs() { return lingerMs; } public void setLingerMs(int lingerMs) { this.lingerMs = lingerMs; } public int getBufferMemory() { return bufferMemory; } public void setBufferMemory(int bufferMemory) { this.bufferMemory = bufferMemory; } } 

测试

消费者处理实现ConsumerHandlerImpl类

public class ConsumerHandlerImpl implements ConsumerHandler{ /** * 处理收到的消息 * @param topic 收到消息的topic名称 * @param partition 收到消息的partition内容 * @param offset 收到消息在队列中的编号 * @param value 收到的消息 */ public void processObject(String topic,int partition,long offset,String value) { System.out.println(topic+"从kafka接收"+partition+"到"+offset+"的消息是："+value); } /** * 获取跨步 * @param topic 接受消息的topic * @param partition 接受消息的partition * @return 当前topic,partition下的seek */ public long getSeek(String topic , int partition) { return 1; } }

main方法类

public class AppResourceTest{ public static void main(String[] args){ BeanDefinitionRegistry reg=new DefaultListableBeanFactory(); PropertiesBeanDefinitionReader reader=new PropertiesBeanDefinitionReader(reg); reader.loadBeanDefinitions(new ClassPathResource("resources/kafka-consumer.properties")); reader.loadBeanDefinitions(new ClassPathResource("resources/kafka-producer.properties")); BeanFactory factory=(BeanFactory)reg; ConsumerConfig consumerConfig=(ConsumerConfig)factory.getBean("consumerConfig"); System.out.println(consumerConfig.getPollTime()); ProducerConfig producerConfig=(ProducerConfig)factory.getBean("producerConfig"); System.out.println(producerConfig.getBatchSize()); Producer producer = new Producer(producerConfig); producer.sendMessage(producerConfig.getTopic(),"s4335453453454"); producer.close(); System.out.println("consumer"); Consumer consumer = new Consumer(new ConsumerHandlerImpl(),consumerConfig); try{ Thread.currentThread(); Thread.sleep(10000); }catch(Exception e){ e.printStackTrace(); } } } 

运行结果