关于kafka-clients JAVA API的基本使用

首先老规矩, 引入maven依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>1.0.0</version>
</dependency>

关于kafka-clients的消息生产者: 

@Slf4j
public class KafkaProducerClient {

    public static void pushMsg(String msg) throws Exception {
        Properties props = new Properties();
        props.put("bootstrap.servers", KafkaConstant.KAFKA_SERVER_ADDRESS);
        props.put("acks", "0");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer producer = new KafkaProducer(props);
        ProducerRecord<String, String> record = new ProducerRecord<>(KafkaConstant.KAFKA_TOPIC_NAME, 0, "123", msg);
        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception e) {
                if (e != null) {
                    e.printStackTrace();
                }
                log.info("pushMsg of msg: {}, metadata: {}", msg, metadata);
            }
        });
        producer.close();
    }
    
}

关于kafka-clients的消息消费者

@Slf4j
public class KafkaConsumerClient extends Thread {
    
    private KafkaConsumerClient() {
    }

    /**
     * 初始化consumer
     */
    public void initKafkaConsumer () {
        log.info("init Kafka Consumer");
        new KafkaConsumerClient().start();
    }
    
    @Override
    public void run() {
        Properties props = new Properties();

        props.put("bootstrap.servers", KafkaConstant.KAFKA_SERVER_ADDRESS);
        props.put("group.id", "1");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        Consumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(KafkaConstant.KAFKA_TOPIC_NAME));
        consumer.seekToBeginning(new ArrayList<>());

        // ===== 拿到所有的topic ===== //
        Map<String, List<PartitionInfo>> listTopics = consumer.listTopics();
        Set<Map.Entry<String, List<PartitionInfo>>> entries = listTopics.entrySet();

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000 * 60);
            for(ConsumerRecord<String, String> record : records) {
                System.out.println("[fetched from partition " + record.partition() + ", offset: " + record.offset() + ", message: " + record.value() + "]");
            }
        }
    }
}

我们需要在项目启动的时候将消费者consumer启动起来

<bean id="initKafkaConsumer" class="com.dywl.zhoushan.kafka.KafkaConsumerClient" init-method="initKafkaConsumer"></bean>

然后调用生产者producer时, 消费者consumer就能拿到消息

如: 

@Override
public void pushMsgById(Long id) throws Exception {
    User user = new User();
    user.setId(id);
    user.setUsername("test11111111");
    user.setPassword("test22222222");
    String str = JsonUtil.toCompactJsonString(user);
    log.info("pushMsgById is user: {}", str);
    KafkaProducerClient.pushMsg(str);
}

得到结果: