Spring boot集成kafka生产者发布消息事件

系统管理员 2020-11-14 17:39 681阅读 0赞

## Spring boot集成kafka ##

1、引入pom文件，版本的可以自己去maven中找

<dependency>
    			<groupId>org.springframework.kafka</groupId>
    			<artifactId>spring-kafka</artifactId>
    			<version>1.2.2.RELEASE</version>
    		</dependency>

2、然后在配置文件中放入你kafka的一些配置

kafka.producer.servers=172.16.100.8:9092
    kafka.producer.retries=0
    kafka.producer.batch.size=16384
    kafka.producer.linger=1
    kafka.producer.buffer.memory=33554432

3、然后增加kafka 的producer的配置类

@Configuration
    @EnableKafka
    public class KafkaProducerConfig {
    
    	@Value("${kafka.producer.servers}")
    	private String servers;
    	@Value("${kafka.producer.retries}")
    	private int retries;
    	@Value("${kafka.producer.batch.size}")
    	private int batchSize;
    	@Value("${kafka.producer.linger}")
    	private int linger;
    	@Value("${kafka.producer.buffer.memory}")
    	private int bufferMemory;
    
    
    	public Map<String, Object> producerConfigs() {
    		Map<String, Object> props = new HashMap<>();
    		props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
    		props.put(ProducerConfig.RETRIES_CONFIG, retries);
    		props.put(ProducerConfig.ACKS_CONFIG, "all");
    		props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
    		props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
    		props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
    		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    		return props;
    	}
    
    
    	public ProducerFactory<String, String> producerFactory() {
    		return new DefaultKafkaProducerFactory<>(producerConfigs());
    	}
    
    	@Bean
    	public KafkaProducer<String, String> kafkaProducer(){
    		return new KafkaProducer<String, String>(producerConfigs(), new StringSerializer(), new StringSerializer());
    	}
    
    	@Bean
    	public KafkaTemplate<String, String> kafkaTemplate() {
    		return new KafkaTemplate<String, String>(producerFactory());
    	}
    }

这里我是配置了 kafkaProducer 和 kafkaTemplate两个。这两个都可以使用。因为我习惯用 kafkaProducer。

5、测试一下

@Autowired
    	private KafkaProducer<String, String> kafkaProducer;
    
    	private void publish(String topic, Event event) {
    		if (StringUtils.isEmpty(topic)) {
    			throw new RuntimeException("topic can not be empty!");
    		}
    		if (event == null) {
    			throw new NullPointerException("event can not be null!");
    		}
    		String json = JSON.toJSONString(event);
    		ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, json);
    //		kafkaTemplate.send(topic, json);
    		kafkaProducer.send(record);
    	}

这是我写的一个utils，试了一下是可以的。Event的话是一个实体类！

public class Event {
    
    	/** 事件类型 */
    	private String type;
    
    	/** 事件唯一标识 */
    	private String id;
    
    	/** 事件时间 */
    	private Date time;
    
    	/** 事件负载 */
    	private String playloadJson;
    
    	/**
    	 * 构造方法
    	 */
    	public Event() {
    
    	}
    
    	/**
    	 * 构造方法
    	 * 
    	 * @param type
    	 */
    	public Event(String type) {
    		this.type = type;
    		this.id = UUID.randomUUID().toString();
    		this.time = new Date();
    	}
    
    	/**
    	 * 取得事件负载
    	 * 
    	 * @param clazz
    	 * @return
    	 */
    	public <T> T getPayload(Class<T> clazz) {
    		if (playloadJson != null) {
    			return JSON.parseObject(playloadJson, clazz);
    		}
    		return null;
    	}
    
    	/**
    	 * 取得事件负载
    	 * 
    	 * @param typeReference
    	 * @return
    	 */
    	public <T> T getPayload(TypeReference<T> typeReference) {
    		if (playloadJson != null) {
    			return JSON.parseObject(playloadJson, typeReference);
    		}
    		return null;
    	}
    
    	/**
    	 * 取得事件负载
    	 * 
    	 * @param clazz
    	 * @return
    	 */
    	public <T> T getPayload(Class<T> clazz, T defaultValue) {
    		if (playloadJson != null) {
    			return JSON.parseObject(playloadJson, clazz);
    		}
    		return defaultValue;
    	}
    
    	/**
    	 * 取得事件负载
    	 * 
    	 * @param typeReference
    	 * @return
    	 */
    	public <T> T getPayload(TypeReference<T> typeReference, T defaultValue) {
    		if (playloadJson != null) {
    			return JSON.parseObject(playloadJson, typeReference);
    		}
    		return defaultValue;
    	}
    
    	/**
    	 * @param playload
    	 *            the playload to set
    	 */
    	public void setPayload(Object playload) {
    		this.playloadJson = JSON.toJSONString(playload, SerializerFeature.DisableCircularReferenceDetect);
    	}
    }

还有一些get和 set方法我没有复制出来。