对接的时候或者处在不同的网络环境下比如在互联网和政务外网的分布部署服务的时候我们需要对接多台kafka来达到我们的业务需求那么当kafka存在多数据源的情况就与单机的情况有所不同。

implementation

org.springframework.kafka:spring-kafka:2.8.2配置

如果是单机的kafka我们直接通过springboot自动配置的就可以使用例如在yml里面直接引用

spring:kafka:producer:key-serializer:

org.apache.kafka.common.serialization.StringSerializervalue-serializer:

org.apache.kafka.common.serialization.StringSerializerconsumer:key-deserializer:

org.apache.kafka.common.serialization.StringDeserializervalue-deserializer:

org.apache.kafka.common.serialization.StringDeserializerbootstrap-servers:

server001.bbd:9092在使用的时候直接注入,然后就可以使用里面的方法了

Resourceprivate

本篇文章主要讲的是在多数据源下的使用和单机的有所不同我也看了网上的一些博客但是当我去按照网上的配置的时候总是会报错

kafakTemplate这个bean找不到,所以没办法只有按照springboot自动配置里面的来改

package

org.springframework.beans.factory.ObjectProvider;

import

org.springframework.beans.factory.annotation.Qualifier;

import

org.springframework.boot.autoconfigure.condition.ConditionalOnClass;

import

org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;

import

org.springframework.boot.autoconfigure.kafka.DefaultKafkaConsumerFactoryCustomizer;

import

org.springframework.boot.autoconfigure.kafka.DefaultKafkaProducerFactoryCustomizer;

import

org.springframework.boot.autoconfigure.kafka.KafkaProperties;

import

org.springframework.boot.context.properties.EnableConfigurationProperties;

import

org.springframework.context.annotation.Bean;

import

org.springframework.context.annotation.Configuration;

import

org.springframework.context.annotation.Primary;

import

org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;

import

org.springframework.kafka.config.KafkaListenerContainerFactory;

import

org.springframework.kafka.core.*;

import

org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import

org.springframework.kafka.security.jaas.KafkaJaasLoginModuleInitializer;

import

org.springframework.kafka.support.LoggingProducerListener;

import

org.springframework.kafka.support.ProducerListener;

import

org.springframework.kafka.support.converter.RecordMessageConverter;

import

org.springframework.kafka.transaction.KafkaTransactionManager;import

java.io.IOException;Configuration(proxyBeanMethods

false)

ConditionalOnClass(KafkaTemplate.class)

EnableConfigurationProperties(KafkaProperties.class)

public

KafkaConfiguration(KafkaProperties

properties,

properties;this.kafkaSecondProperties

kafkaSecondProperties;}Bean(kafkaTemplate)Primarypublic

KafkaTemplate?,

kafkaTemplate(ProducerFactoryObject,

Object

kafkaProducerFactory,ProducerListenerObject,

Object

kafkaProducerListener,ObjectProviderRecordMessageConverter

messageConverter)

KafkaTemplate(kafkaProducerFactory);messageConverter.ifUnique(kafkaTemplate::setMessageConverter);kafkaTemplate.setProducerListener(kafkaProducerListener);kafkaTemplate.setDefaultTopic(this.properties.getTemplate().getDefaultTopic());return

kafkaTemplate;}Bean(kafkaSecondTemplate)public

KafkaTemplate?,

kafkaSecondTemplate(Qualifier(kafkaSecondProducerFactory)

Object

kafkaProducerFactory,Qualifier(kafkaSecondProducerListener)

Object

kafkaProducerListener,ObjectProviderRecordMessageConverter

messageConverter)

KafkaTemplate(kafkaProducerFactory);messageConverter.ifUnique(kafkaTemplate::setMessageConverter);kafkaTemplate.setProducerListener(kafkaProducerListener);kafkaTemplate.setDefaultTopic(this.properties.getTemplate().getDefaultTopic());return

kafkaTemplate;}Bean(kafkaProducerListener)Primarypublic

Object

LoggingProducerListener();}Bean(kafkaSecondProducerListener)public

Object

LoggingProducerListener();}Bean(kafkaConsumerFactory)Primarypublic

Object

kafkaConsumerFactory(ObjectProviderDefaultKafkaConsumerFactoryCustomizer

customizers)

{DefaultKafkaConsumerFactoryObject,

Object

DefaultKafkaConsumerFactory(this.properties.buildConsumerProperties());customizers.orderedStream().forEach((customizer)

customizer.customize(factory));return

factory;}Bean(kafkaSecondConsumerFactory)public

Object

kafkaSecondConsumerFactory(ObjectProviderDefaultKafkaConsumerFactoryCustomizer

customizers)

{DefaultKafkaConsumerFactoryObject,

Object

DefaultKafkaConsumerFactory(this.kafkaSecondProperties.buildConsumerProperties());customizers.orderedStream().forEach((customizer)

customizer.customize(factory));return

factory;}Bean(zwKafkaContainerFactory)KafkaListenerContainerFactoryConcurrentMessageListenerContainerInteger,

String

zwKafkaContainerFactory(Qualifier(value

Object

{ConcurrentKafkaListenerContainerFactoryInteger,

String

ConcurrentKafkaListenerContainerFactory();factory.setConsumerFactory(kafkaSecondConsumerFactory);factory.setConcurrency(3);factory.getContainerProperties().setPollTimeout(3000);return

factory;}Bean(kafkaProducerFactory)Primarypublic

Object

kafkaProducerFactory(ObjectProviderDefaultKafkaProducerFactoryCustomizer

customizers)

{DefaultKafkaProducerFactoryObject,

Object

DefaultKafkaProducerFactory(this.properties.buildProducerProperties());String

transactionIdPrefix

this.properties.getProducer().getTransactionIdPrefix();if

(transactionIdPrefix

{factory.setTransactionIdPrefix(transactionIdPrefix);}customizers.orderedStream().forEach((customizer)

customizer.customize(factory));return

factory;}Bean(kafkaSecondProducerFactory)public

Object

kafkaSecondProducerFactory(ObjectProviderDefaultKafkaProducerFactoryCustomizer

customizers)

{DefaultKafkaProducerFactoryObject,

Object

DefaultKafkaProducerFactory(this.kafkaSecondProperties.buildProducerProperties());String

transactionIdPrefix

this.kafkaSecondProperties.getProducer().getTransactionIdPrefix();if

(transactionIdPrefix

{factory.setTransactionIdPrefix(transactionIdPrefix);}customizers.orderedStream().forEach((customizer)

customizer.customize(factory));return

factory;}BeanConditionalOnProperty(name

spring.kafka.producer.transaction-id-prefix)public

kafkaTransactionManager(ProducerFactory?,

producerFactory)

KafkaTransactionManager(producerFactory);}BeanConditionalOnProperty(name

spring.kafka.jaas.enabled)public

KafkaJaasLoginModuleInitializer

throws

{KafkaJaasLoginModuleInitializer

jaas

KafkaJaasLoginModuleInitializer();KafkaProperties.Jaas

jaasProperties

(jaasProperties.getControlFlag()

null)

{jaas.setControlFlag(jaasProperties.getControlFlag());}if

(jaasProperties.getLoginModule()

null)

{jaas.setLoginModule(jaasProperties.getLoginModule());}jaas.setOptions(jaasProperties.getOptions());return

jaas;}Bean(kafkaAdmin)Primarypublic

KafkaAdmin

KafkaAdmin(this.properties.buildAdminProperties());kafkaAdmin.setFatalIfBrokerNotAvailable(this.properties.getAdmin().isFailFast());return

kafkaAdmin;}}

com.alibaba.fastjson.JSONObject;import

com.ddb.zggz.config.ApplicationConfiguration;

import

org.springframework.beans.factory.annotation.Autowired;

import

org.springframework.beans.factory.annotation.Qualifier;

import

org.springframework.kafka.core.KafkaTemplate;

import

org.springframework.kafka.support.SendResult;

import

org.springframework.stereotype.Component;

import

org.springframework.util.concurrent.ListenableFuture;

import

org.springframework.util.concurrent.ListenableFutureCallback;import

javax.annotation.Resource;Component

Slf4j

kafkaSecondTemplate;Resourceprivate

KafkaTemplateString,

{ListenableFutureSendResultString,

String

(zw.equals(configuration.getEnvironment())){sendResultListenableFuture

kafkaSecondTemplate.send(configuration.getPushTopic(),

(net.equals(configuration.getEnvironment())){sendResultListenableFuture

kafkaTemplate.send(configuration.getPushTopic(),

null){throw

IllegalArgumentException(kakfa发送消息失败);}sendResultListenableFuture.addCallback(new

ListenableFutureCallbackSendResultString,

String()

{log.error(kafka发送的message报错发送数据{},

void

{log.info(kafka发送的message成功发送数据{},

param);}});}}消费者

com.alibaba.fastjson.JSONObject;import

com.ddb.zggz.config.ApplicationConfiguration;

import

com.ddb.zggz.model.dto.ApprovalDTO;

import

com.ddb.zggz.param.OffShelfParam;

import

com.ddb.zggz.service.GzApprovalService;

import

com.ddb.zggz.service.GzServiceService;

import

org.apache.commons.lang3.StringUtils;

import

org.apache.kafka.clients.consumer.ConsumerRecord;

import

org.springframework.beans.factory.annotation.Autowired;

import

org.springframework.kafka.annotation.DltHandler;

import

org.springframework.kafka.annotation.KafkaListener;

import

org.springframework.kafka.annotation.RetryableTopic;

import

org.springframework.retry.annotation.Backoff;

import

org.springframework.stereotype.Component;

import

org.springframework.util.ObjectUtils;import

java.util.ArrayList;

gzApprovalService;Autowiredprivate

GzServiceService

gzServiceService;KafkaListener(topics

${application.config.push-topic},

groupId

zwKafkaContainerFactory)RetryableTopic(include

Backoff(delay

consumerRecord.value();PushParam

pushParam

(version-approval.equals(pushParam.getEvent()))

{ApprovalDTO

JSONObject.parseObject(JSONObject.toJSONString(pushParam.getData()),

ApprovalDTO.class);gzApprovalService.approval(approvalDTO);}//服务下架if

(pushParam.getEvent().equals(server-OffShelf-gzt))

{OffShelfParam

JSONObject.parseObject(JSONObject.toJSONString(pushParam.getData()),

OffShelfParam.class);gzServiceService.offShelfV1(offShelfParam.getReason(),

null,

offShelfParam.getVersion());}}DltHandlerpublic

void

com.alibaba.fastjson.annotation.JSONField;

import

com.ddb.zggz.model.GzH5VersionManage;

import

com.ddb.zggz.model.dto.ApprovalDTO;

import

com.ddb.zggz.param.OffShelfParam;

import

com.ddb.zggz.param.PublishParam;

import

com.ddb.zggz.param.ReviewAndRollback;

import

com.fasterxml.jackson.annotation.JsonFormat;

import

com.fasterxml.jackson.databind.annotation.JsonDeserialize;

import

com.fasterxml.jackson.databind.annotation.JsonSerialize;

import

com.fasterxml.jackson.datatype.jsr310.deser.LocalDateTimeDeserializer;

import

com.fasterxml.jackson.datatype.jsr310.ser.LocalDateTimeSerializer;

import

LocalDateTimeSerializer.class)JsonDeserialize(using

LocalDateTimeDeserializer.class)JSONField(format

yyyy-MM-dd

toKafkaVersion(GzH5VersionManage

gzH5VersionManage)

PushParam();pushParam.setData(gzH5VersionManage);pushParam.setEvent(save-version);return

pushParam;}/***

PushParam();pushParam.setData(gzService);pushParam.setEvent(save-server);return

pushParam;}