Java实现Kafka生产者和消费者的示例

Kafka简介

Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka的目标是为处理实时数据提供一个统一、高吞吐、低延迟的平台。

[图片上传失败...(image-620664-1652680507922)]

方式一：kafka-clients

引入依赖

在pom.xml文件中，引入kafka-clients依赖：

<pre class="prettyprint hljs xml" style="padding: 0.5em; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; color: rgb(68, 68, 68); border-radius: 4px; display: block; margin: 0px 0px 1.5em; font-size: 14px; line-height: 1.5em; word-break: break-all; overflow-wrap: break-word; white-space: pre; background-color: rgb(246, 246, 246); border: none; overflow-x: auto; font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: normal; orphans: 2; text-align: start; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; text-decoration-thickness: initial; text-decoration-style: initial; text-decoration-color: initial;"><dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.3.1</version>
</dependency></pre>

生产者

创建一个KafkaProducer的生产者实例：

<pre class="prettyprint hljs dart" style="padding: 0.5em; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; color: rgb(68, 68, 68); border-radius: 4px; display: block; margin: 0px 0px 1.5em; font-size: 14px; line-height: 1.5em; word-break: break-all; overflow-wrap: break-word; white-space: pre; background-color: rgb(246, 246, 246); border: none; overflow-x: auto; font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: normal; orphans: 2; text-align: start; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; text-decoration-thickness: initial; text-decoration-style: initial; text-decoration-color: initial;">@Configuration
public class Config {

public final static String bootstrapServers = "127.0.0.1:9092";

@Bean(destroyMethod = "close")
public KafkaProducer<String, String> kafkaProducer() {
    Properties props = new Properties();
    //设置Kafka服务器地址
    props.put("bootstrap.servers", bootstrapServers);
    //设置数据key的序列化处理类
    props.put("key.serializer", StringSerializer.class.getName());
    //设置数据value的序列化处理类
    props.put("value.serializer", StringSerializer.class.getName());
    KafkaProducer<String, String> producer = new KafkaProducer<>(props);
    return producer;
}

}</pre>

在Controller中进行使用：

<pre class="prettyprint hljs dart" style="padding: 0.5em; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; color: rgb(68, 68, 68); border-radius: 4px; display: block; margin: 0px 0px 1.5em; font-size: 14px; line-height: 1.5em; word-break: break-all; overflow-wrap: break-word; white-space: pre; background-color: rgb(246, 246, 246); border: none; overflow-x: auto; font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: normal; orphans: 2; text-align: start; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; text-decoration-thickness: initial; text-decoration-style: initial; text-decoration-color: initial;">@RestController
@Slf4j
public class Controller {

@Autowired
private KafkaProducer<String, String> kafkaProducer;

@RequestMapping("/kafkaClientsSend")
public String send() {
    String uuid = UUID.randomUUID().toString();
    RecordMetadata recordMetadata = null;
    try {
        //将消息发送到Kafka服务器的名称为“one-more-topic”的Topic中
        recordMetadata = kafkaProducer.send(new ProducerRecord<>("one-more-topic", uuid)).get();
        log.info("recordMetadata: {}", recordMetadata);
        log.info("uuid: {}", uuid);
    } catch (Exception e) {
        log.error("send fail, uuid: {}", uuid, e);
    }
    return uuid;
}

}</pre>

消费者

创建一个KafkaConsumer的消费者实例：

<pre class="prettyprint hljs dart" style="padding: 0.5em; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; color: rgb(68, 68, 68); border-radius: 4px; display: block; margin: 0px 0px 1.5em; font-size: 14px; line-height: 1.5em; word-break: break-all; overflow-wrap: break-word; white-space: pre; background-color: rgb(246, 246, 246); border: none; overflow-x: auto; font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: normal; orphans: 2; text-align: start; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; text-decoration-thickness: initial; text-decoration-style: initial; text-decoration-color: initial;">@Configuration
public class Config {

public final static String groupId = "kafka-clients-group";
public final static String bootstrapServers = "127.0.0.1:9092";

@Bean(destroyMethod = "close")
public KafkaConsumer<String, String> kafkaConsumer() {
    Properties props = new Properties();
    //设置Kafka服务器地址
    props.put("bootstrap.servers", bootstrapServers);
    //设置消费组
    props.put("group.id", groupId);
    //设置数据key的反序列化处理类
    props.put("key.deserializer", StringDeserializer.class.getName());
    //设置数据value的反序列化处理类
    props.put("value.deserializer", StringDeserializer.class.getName());
    props.put("enable.auto.commit", "true");
    props.put("auto.commit.interval.ms", "1000");
    props.put("session.timeout.ms", "30000");
    KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
    //订阅名称为“one-more-topic”的Topic的消息
    kafkaConsumer.subscribe(Arrays.asList("one-more-topic"));
    return kafkaConsumer;
}

}</pre>

在Controller中进行使用：

<pre class="prettyprint hljs dart" style="padding: 0.5em; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; color: rgb(68, 68, 68); border-radius: 4px; display: block; margin: 0px 0px 1.5em; font-size: 14px; line-height: 1.5em; word-break: break-all; overflow-wrap: break-word; white-space: pre; background-color: rgb(246, 246, 246); border: none; overflow-x: auto; font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: normal; orphans: 2; text-align: start; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; text-decoration-thickness: initial; text-decoration-style: initial; text-decoration-color: initial;">@RestController
@Slf4j
public class Controller {

@Autowired
private KafkaConsumer<String, String> kafkaConsumer;

@RequestMapping("/receive")
public List<String> receive() {
    ////从Kafka服务器中的名称为“one-more-topic”的Topic中消费消息
    ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(1));
    List<String> messages = new ArrayList<>(records.count());
    for (ConsumerRecord<String, String> record : records.records("one-more-topic")) {
        String message = record.value();
        log.info("message: {}", message);
        messages.add(message);
    }
    return messages;
}

}</pre>

方式二：spring-kafka

使用kafka-clients需要我们自己创建生产者或者消费者的bean，如果我们的项目基于SpringBoot构建，那么使用spring-kafka就方便多了。

引入依赖

在pom.xml文件中，引入spring-kafka依赖：

<pre class="prettyprint hljs xml" style="padding: 0.5em; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; color: rgb(68, 68, 68); border-radius: 4px; display: block; margin: 0px 0px 1.5em; font-size: 14px; line-height: 1.5em; word-break: break-all; overflow-wrap: break-word; white-space: pre; background-color: rgb(246, 246, 246); border: none; overflow-x: auto; font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: normal; orphans: 2; text-align: start; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; text-decoration-thickness: initial; text-decoration-style: initial; text-decoration-color: initial;"><dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
<version>2.3.12.RELEASE</version>
</dependency></pre>

生产者

在application.yml文件中增加配置：

<pre class="prettyprint hljs css" style="padding: 0.5em; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; color: rgb(68, 68, 68); border-radius: 4px; display: block; margin: 0px 0px 1.5em; font-size: 14px; line-height: 1.5em; word-break: break-all; overflow-wrap: break-word; white-space: pre; background-color: rgb(246, 246, 246); border: none; overflow-x: auto; font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: normal; orphans: 2; text-align: start; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; text-decoration-thickness: initial; text-decoration-style: initial; text-decoration-color: initial;">spring:
kafka:
#Kafka服务器地址
bootstrap-servers: 127.0.0.1:9092
producer:
#设置数据value的序列化处理类
value-serializer: org.apache.kafka.common.serialization.StringSerializer</pre>

在Controller中注入KafkaTemplate就可以直接使用了，代码如下：

<pre class="prettyprint hljs kotlin" style="padding: 0.5em; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; color: rgb(68, 68, 68); border-radius: 4px; display: block; margin: 0px 0px 1.5em; font-size: 14px; line-height: 1.5em; word-break: break-all; overflow-wrap: break-word; white-space: pre; background-color: rgb(246, 246, 246); border: none; overflow-x: auto; font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: normal; orphans: 2; text-align: start; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; text-decoration-thickness: initial; text-decoration-style: initial; text-decoration-color: initial;">@RestController
@Slf4j
public class Controller {

@Autowired
private KafkaTemplate<String, String> template;

@RequestMapping("/springKafkaSend")
public String send() {
    String uuid = UUID.randomUUID().toString();
    //将消息发送到Kafka服务器的名称为“one-more-topic”的Topic中
    this.template.send("one-more-topic", uuid);
    log.info("uuid: {}", uuid);
    return uuid;
}

}</pre>

消费者

在application.yml文件中增加配置：

<pre class="prettyprint hljs css" style="padding: 0.5em; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; color: rgb(68, 68, 68); border-radius: 4px; display: block; margin: 0px 0px 1.5em; font-size: 14px; line-height: 1.5em; word-break: break-all; overflow-wrap: break-word; white-space: pre; background-color: rgb(246, 246, 246); border: none; overflow-x: auto; font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: normal; orphans: 2; text-align: start; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; text-decoration-thickness: initial; text-decoration-style: initial; text-decoration-color: initial;">spring:
kafka:
#Kafka服务器地址
bootstrap-servers: 127.0.0.1:9092
consumer:
#设置数据value的反序列化处理类
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer</pre>

创建一个可以被Spring框架扫描到的类，并且在方法上加上@KafkaListener注解，就可以消费消息了，代码如下：

<pre class="prettyprint hljs dart" style="padding: 0.5em; font-family: Menlo, Monaco, Consolas, "Courier New", monospace; color: rgb(68, 68, 68); border-radius: 4px; display: block; margin: 0px 0px 1.5em; font-size: 14px; line-height: 1.5em; word-break: break-all; overflow-wrap: break-word; white-space: pre; background-color: rgb(246, 246, 246); border: none; overflow-x: auto; font-style: normal; font-variant-ligatures: normal; font-variant-caps: normal; font-weight: 400; letter-spacing: normal; orphans: 2; text-align: start; text-indent: 0px; text-transform: none; widows: 2; word-spacing: 0px; -webkit-text-stroke-width: 0px; text-decoration-thickness: initial; text-decoration-style: initial; text-decoration-color: initial;">@Component
@Slf4j
public class Receiver {

@KafkaListener(topics = "one-more-topic", groupId = "spring-kafka-group")
public void listen(ConsumerRecord<?, ?> record) {
    Optional<?> kafkaMessage = Optional.ofNullable(record.value());
    if (kafkaMessage.isPresent()) {
        String message = (String) kafkaMessage.get();
        log.info("message: {}", message);
    }
}

}</pre>