springboot配置双kafka使用spring boot 2.0.8.release 版本
引入maven kafka jar、准备两个kafka；<dependency> <groupid>org.springframework.kafka</groupid> <artifactid>spring-kafka</artifactid></dependency>
配置yml配置文件spring: kafka: bootstrap-servers: 180.167.180.242:9092 #kafka的访问地址，多个用","隔开 consumer: enable-auto-commit: true group-id: kafka #群组id outkafka: bootstrap-servers: localhost:9092 #kafka的访问地址，多个用","隔开 consumer: enable-auto-commit: true group-id: kafka_1 #群组id
配置kafkaconfig类import java.util.hashmap;import java.util.map; import org.apache.kafka.clients.consumer.consumerconfig;import org.apache.kafka.clients.producer.producerconfig;import org.apache.kafka.common.serialization.stringdeserializer;import org.apache.kafka.common.serialization.stringserializer;import org.springframework.beans.factory.annotation.value;import org.springframework.context.annotation.bean;import org.springframework.context.annotation.configuration;import org.springframework.context.annotation.primary;import org.springframework.kafka.annotation.enablekafka;import org.springframework.kafka.config.concurrentkafkalistenercontainerfactory;import org.springframework.kafka.config.kafkalistenercontainerfactory;import org.springframework.kafka.core.consumerfactory;import org.springframework.kafka.core.defaultkafkaconsumerfactory;import org.springframework.kafka.core.defaultkafkaproducerfactory;import org.springframework.kafka.core.kafkatemplate;import org.springframework.kafka.core.producerfactory;import org.springframework.kafka.listener.concurrentmessagelistenercontainer; @configuration@enablekafkapublic class kafkaconfig { @value("${spring.kafka.bootstrap-servers}") private string innerservers; @value("${spring.kafka.consumer.group-id}") private string innergroupid; @value("${spring.kafka.consumer.enable-auto-commit}") private string innerenableautocommit; @bean @primary//理解为默认优先选择当前容器下的消费者工厂 kafkalistenercontainerfactory<concurrentmessagelistenercontainer<integer, string>> kafkalistenercontainerfactory() { concurrentkafkalistenercontainerfactory<integer, string> factory = new concurrentkafkalistenercontainerfactory<>(); factory.setconsumerfactory(consumerfactory()); factory.setconcurrency(3); factory.getcontainerproperties().setpolltimeout(3000); return factory; } @bean//第一个消费者工厂的bean public consumerfactory<integer, string> consumerfactory() { return new defaultkafkaconsumerfactory<>(consumerconfigs()); } @bean public map<string, object> consumerconfigs() { map<string, object> props = new hashmap<>(); props.put(consumerconfig.bootstrap_servers_config, innerservers); props.put(consumerconfig.group_id_config, innergroupid); props.put(consumerconfig.enable_auto_commit_config, innerenableautocommit);// props.put(consumerconfig.auto_commit_interval_ms_config, "100");// props.put(consumerconfig.session_timeout_ms_config, "15000"); props.put(consumerconfig.key_deserializer_class_config, stringdeserializer.class); props.put(consumerconfig.value_deserializer_class_config, stringdeserializer.class); return props; } @bean //生产者工厂配置 public producerfactory<string, string> producerfactory() { return new defaultkafkaproducerfactory<>(senderprops()); } @bean //kafka发送消息模板 public kafkatemplate<string, string> kafkatemplate() { return new kafkatemplate<string, string>(producerfactory()); } /** * 生产者配置方法 * * 生产者有三个必选属性 * <p> * 1.bootstrap.servers broker地址清单，清单不要包含所有的broker地址， * 生产者会从给定的broker里查找到其他broker的信息。不过建议至少提供两个broker信息，一旦 其中一个宕机，生产者仍能能够连接到集群上。 * </p> * <p> * 2.key.serializer broker希望接收到的消息的键和值都是字节数组。 生产者用对应的类把键对象序列化成字节数组。 * </p> * <p> * 3.value.serializer 值得序列化方式 * </p> * * * @return */ private map<string, object> senderprops() { map<string, object> props = new hashmap<>(); props.put(producerconfig.bootstrap_servers_config, innerservers); /** * 当从broker接收到的是临时可恢复的异常时，生产者会向broker重发消息，但是不能无限 * 制重发，如果重发次数达到限制值，生产者将不会重试并返回错误。 * 通过retries属性设置。默认情况下生产者会在重试后等待100ms，可以通过 retries.backoff.ms属性进行修改 */ props.put(producerconfig.retries_config, 0); /** * 在考虑完成请求之前，生产者要求leader收到的确认数量。这可以控制发送记录的持久性。允许以下设置： * <ul> * <li> * <code> acks = 0 </ code>如果设置为零，则生产者将不会等待来自服务器的任何确认。该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且 * <code>retries </ code>配置将不会生效（因为客户端通常不会知道任何故障）。为每条记录返回的偏移量始终设置为-1。 * <li> <code> acks = 1 </code> * 这意味着leader会将记录写入其本地日志，但无需等待所有follower的完全确认即可做出回应。在这种情况下， * 如果leader在确认记录后立即失败但在关注者复制之前，则记录将丢失。 * <li><code> acks = all </code> * 这意味着leader将等待完整的同步副本集以确认记录。这保证了只要至少一个同步副本仍然存活，记录就不会丢失。这是最强有力的保证。 * 这相当于acks = -1设置 */ props.put(producerconfig.acks_config, "1"); /** * 当有多条消息要被发送到统一分区是，生产者会把他们放到统一批里。kafka通过批次的概念来 提高吞吐量，但是也会在增加延迟。 */ // 以下配置当缓存数量达到16kb，就会触发网络请求，发送消息// props.put(producerconfig.batch_size_config, 16384); // 每条消息在缓存中的最长时间，如果超过这个时间就会忽略batch.size的限制，由客户端立即将消息发送出去// props.put(producerconfig.linger_ms_config, 1);// props.put(producerconfig.buffer_memory_config, 33554432); props.put(producerconfig.key_serializer_class_config, stringserializer.class); props.put(producerconfig.value_serializer_class_config, stringserializer.class); return props; } @value("${spring.outkafka.bootstrap-servers}") private string outservers; @value("${spring.outkafka.consumer.group-id}") private string outgroupid; @value("${spring.outkafka.consumer.enable-auto-commit}") private string outenableautocommit; static { } /** * 连接第二个kafka集群的配置 */ @bean kafkalistenercontainerfactory<concurrentmessagelistenercontainer<integer, string>> kafkalistenercontainerfactoryoutschedule() { concurrentkafkalistenercontainerfactory<integer, string> factory = new concurrentkafkalistenercontainerfactory<>(); factory.setconsumerfactory(consumerfactoryoutschedule()); factory.setconcurrency(3); factory.getcontainerproperties().setpolltimeout(3000); return factory; } @bean public consumerfactory<integer, string> consumerfactoryoutschedule() { return new defaultkafkaconsumerfactory<>(consumerconfigsoutschedule()); } /** * 连接第二个集群的消费者配置 */ @bean public map<string, object> consumerconfigsoutschedule() { map<string, object> props = new hashmap<>(); props.put(consumerconfig.bootstrap_servers_config, outservers); props.put(consumerconfig.group_id_config, outgroupid); props.put(consumerconfig.enable_auto_commit_config, outenableautocommit); props.put(consumerconfig.key_deserializer_class_config, stringdeserializer.class); props.put(consumerconfig.value_deserializer_class_config, stringdeserializer.class); return props; } @bean //生产者工厂配置 public producerfactory<string, string> produceroutfactory() { return new defaultkafkaproducerfactory<>(senderoutprops()); } @bean //kafka发送消息模板 public kafkatemplate<string, string> kafkaouttemplate() { return new kafkatemplate<string, string>(produceroutfactory()); } /** * 生产者配置方法 * * 生产者有三个必选属性 * <p> * 1.bootstrap.servers broker地址清单，清单不要包含所有的broker地址， * 生产者会从给定的broker里查找到其他broker的信息。不过建议至少提供两个broker信息，一旦 其中一个宕机，生产者仍能能够连接到集群上。 * </p> * <p> * 2.key.serializer broker希望接收到的消息的键和值都是字节数组。 生产者用对应的类把键对象序列化成字节数组。 * </p> * <p> * 3.value.serializer 值得序列化方式 * </p> * * * @return */ private map<string, object> senderoutprops() { map<string, object> props = new hashmap<>(); props.put(producerconfig.bootstrap_servers_config, outservers); /** * 当从broker接收到的是临时可恢复的异常时，生产者会向broker重发消息，但是不能无限 * 制重发，如果重发次数达到限制值，生产者将不会重试并返回错误。 * 通过retries属性设置。默认情况下生产者会在重试后等待100ms，可以通过 retries.backoff.ms属性进行修改 */ props.put(producerconfig.retries_config, 0); /** * 在考虑完成请求之前，生产者要求leader收到的确认数量。这可以控制发送记录的持久性。允许以下设置： * <ul> * <li> * <code> acks = 0 </ code>如果设置为零，则生产者将不会等待来自服务器的任何确认。该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且 * <code>retries </ code>配置将不会生效（因为客户端通常不会知道任何故障）。为每条记录返回的偏移量始终设置为-1。 * <li> <code> acks = 1 </code> * 这意味着leader会将记录写入其本地日志，但无需等待所有follower的完全确认即可做出回应。在这种情况下， * 如果leader在确认记录后立即失败但在关注者复制之前，则记录将丢失。 * <li><code> acks = all </code> * 这意味着leader将等待完整的同步副本集以确认记录。这保证了只要至少一个同步副本仍然存活，记录就不会丢失。这是最强有力的保证。 * 这相当于acks = -1设置 */ props.put(producerconfig.acks_config, "1"); /** * 当有多条消息要被发送到统一分区是，生产者会把他们放到统一批里。kafka通过批次的概念来 提高吞吐量，但是也会在增加延迟。 */ // 以下配置当缓存数量达到16kb，就会触发网络请求，发送消息// props.put(producerconfig.batch_size_config, 16384); // 每条消息在缓存中的最长时间，如果超过这个时间就会忽略batch.size的限制，由客户端立即将消息发送出去// props.put(producerconfig.linger_ms_config, 1);// props.put(producerconfig.buffer_memory_config, 33554432); props.put(producerconfig.key_serializer_class_config, stringserializer.class); props.put(producerconfig.value_serializer_class_config, stringserializer.class); return props; }}
发送工具类mykafkaproducerimport org.springframework.beans.factory.annotation.autowired;import org.springframework.kafka.core.kafkatemplate;import org.springframework.kafka.support.sendresult;import org.springframework.scheduling.annotation.enablescheduling;import org.springframework.stereotype.component;import org.springframework.util.concurrent.listenablefuture; import lombok.extern.slf4j.slf4j; /** * <p> * <b>kafkaproducer description:</b> kafka生产者 * </p> * * @author douzaixing<b>date</b> 2019年7月8日 下午4:09:29 */@component // 这个必须加入容器不然，不会执行@enablescheduling // 这里是为了测试加入定时调度@slf4jpublic class mykafkaproducer { @autowired private kafkatemplate<string, string> kafkatemplate; @autowired private kafkatemplate<string, string> kafkaouttemplate; public listenablefuture<sendresult<string, string>> send(string topic, string key, string json) { listenablefuture<sendresult<string, string>> result = kafkatemplate.send(topic, key, json); log.info("inner kafka send #topic=" + topic + "#key=" + key + "#json=" + json + "#推送成功==========="); return result; } public listenablefuture<sendresult<string, string>> sendout(string topic, string key, string json) { listenablefuture<sendresult<string, string>> result = kafkaouttemplate.send(topic, key, json); log.info("out kafka send #topic=" + topic + "#key=" + key + "#json=" + json + "#推送成功==========="); return result; } }
测试类@slf4j@runwith(springjunit4classrunner.class)@springboottest(classes={oesbcserviceapplication.class})public class morekafkatest { @autowired private mykafkaproducer kafkaproducer; @test public void sendinner() { for (int i = 0; i < 1; i++) { kafkaproducer.send("inner_test", "douzi" + i, "liyuehua" + i); kafkaproducer.sendout("out_test", "douziout" + i, "fanbingbing" + i); } }}
接收类@component@slf4jpublic class kafkaconsumer { @kafkalistener(topics={"inner_test"}, containerfactory="kafkalistenercontainerfactory") public void innerlistener(consumerrecord<string, string> record) { log.info("inner kafka receive #key=" + record.key() + "#value=" + record.value()); } @kafkalistener(topics={"out_test"}, containerfactory="kafkalistenercontainerfactoryoutschedule") public void outlistener(consumerrecord<string, string> record) { log.info("out kafka receive #key=" + record.key() + "#value=" + record.value()); }}
测试结果07-11 12:41:27.811 info  [com.wondertek.oes.bc.service.send.mykafkaproducer] - inner kafka send #topic=inner_test#key=douzi0#json=liyuehua0#推送成功===========
07-11 12:41:27.995 info  [com.wondertek.oes.bc.service.send.kafkaconsumer] - inner kafka receive #key=douzi0#value=liyuehua0
07-11 12:41:28.005 info  [com.wondertek.oes.bc.service.send.mykafkaproducer] - out kafka send #topic=out_test#key=douziout0#json=fanbingbing0#推送成功===========
07-11 12:41:28.013 info  [com.wondertek.oes.bc.service.send.kafkaconsumer] - out kafka receive #key=douziout0#value=fanbingbing0
以上就是springboot怎么配置双kafka的详细内容。