Spring boot集成kafka生产者发布消息事件-蒲公英云

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

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);
    }
}