使用 Kafka 开发 Java 应用程序

导入java。乌蒂尔.哈希地图;

导入java。乌蒂尔.地图;

@Configuration

公共类生产者配置|

私人静态最终字符串KAFKA_BROKER="本地主机：9092";

@Bean

19

  }

@Bean

公共映射<字符串对象>生产者配置（） |

映射<字符串对象>配置=新的哈希Map<> （;

配置。put（生产者 Config.BOOTSTRAP_SERVERS_CONFIGKAFKA_BROKER）;

配置

配置。put（生产者 Config.VALUE_SERIALIZER_CLASS_CONFIG字符串序列器。类）;

返回配置;

  }

@Bean

公共KafkaTemplate<字符串>kafkaTemplate（） |

返回新的KafkaTemplate<>（生产商工厂（））;

  }

导入组织.弹簧框架。网络.绑定。注释。获取映射;

导入组织.弹簧框架。网络.绑定。注释。请求Param;

导入组织.弹簧框架。网络.绑定。注释。休息控制器;

导入java。乌蒂尔.列表;

@RestController

公共类卡夫卡控制器|

公共卡夫卡控制器（卡夫卡模板<字符串，字符串> 模板） ]



14

这个.模板=模板;



15

  }



16





17

@GetMapping（"/卡夫卡/生产"）



18

公共无效产生（@RequestParam字符串消息） |



19

模板。发送（"myTopic"消息）;



20

  }



21












注意：由于您要发送要处理的数据， produce() 因此该方法确实应该是 POST
正如您所看到的，此终结点非常简单。它较早地注入配置的 ， KafkaTemplate 并在向 myTopic 发出请求时向 发送消息 GET /kafka/produce 。
让我们测试一下是否一切按预期工作。main在类中运行方法 JavaKafkaApplication 。要从命令行运行，请执行以下命令：



壳





























xxxxxxxxx

1














1

./mvnw 弹簧启动：运行




















您的服务器应在端口 8080 上运行，并且您已经可以针对它发出 API 请求！
转到您的 Web 浏览器并访问http://localhost:8080/kafka/produce?message=This是我的消息。
使用上述命令进行调用时，应用程序将执行 /kafka/produce 终结点，该终结点向 myTopic Kafka 内的主题发送消息。
但是，您如何知道该命令已成功向主题发送消息？现在，您不会使用应用内的消息，这意味着您无法确定！
幸运的是，有一个简单的方法来创建一个消费者来直接测试。在 bin Kafka 目录的文件夹中，运行以下命令：



壳




























xxxxxxxxxx

1














1

--引导服务器--主题我的专题

















访问http://localhost:8080/kafka/produce?message=This是我在运行 Kafka 使用者的终端中再次看到以下消息的消息：



壳





























xxxxxxxxxx

1












1

这是我的信息











干得好！您可以立即停止此命令。
让我们添加一些 Java 代码来使用应用内的消息，而不是从终端执行。
在 Java 应用程序中使用来自 Kafka 主题的消息
与生产者一样，您需要添加配置以使使用者能够找到 Kafka Broker。
在 src/main/java/com/okta/javakafka/configuration 中创建以下类：



Java





























xxxxxxxxx

1


42












1

导入组织.阿帕奇.卡夫卡.客户端。消费者。消费者协会;



2

导入组织.阿帕奇.卡夫卡.常见。序列化.字符串反序列化器;



3

导入组织.弹簧框架。上下文。注释。豆豆;



4

导入组织.弹簧框架。上下文。注释。配置;



5

导入组织.弹簧框架。卡夫卡.config.并发卡夫卡接收容器工厂;



6

导入组织.弹簧框架。卡夫卡.核心.消费工厂;



7

导入组织.弹簧框架。卡夫卡


默认卡夫卡消费者工厂;


8





9

导入java。乌蒂尔.哈希地图;



10

导入java。乌蒂尔.地图;



11





12

@Configuration



13

公共类消费者配置|



14





15

私人静态最终字符串KAFKA_BROKER="本地主机：9092";



16

私人静态最终字符串GROUP_ID="卡夫卡-沙盒";



17




@Bean



19

公共消费工厂<字符串>消费工厂（） |



20

返回新的DefaultKafka 消费者工厂<>（消费者配置））;



21

  }



22





23

@Bean



24

公共映射<字符串对象>使用者配置（） |



25

映射<字符串对象>配置=新的哈希Map<> （;



26





27

put（消费者Config. BOOTSTRAP_SERVERS_CONFIG，KAFKA_BROKER”



28

配置。put（消费者Config.GROUP_ID_CONFIGGROUP_ID）;



29

配置。put（消费者Config.KEY_DESERIALIZER_CLASS_CONFIG字符串反序列化器。类）;



30

配置。put（消费者Config.VALUE_DESERIALIZER_CLASS_CONFIG字符串反序列化器。类）;



31





32

返回配置;



33

  }



34





35

@Bean



36

37
并发 Kafka Listener 容器工厂<字符串>工厂=新并发 Kafka Listener 容器工厂<> （;


38

工厂。集消费者工厂（消费者工厂（））;



39

返回工厂;



40

  }



41





42

}










上面的代码创建一个工厂，知道如何连接到本地代理。它还会将使用者配置为对 String 键和值进行反序列化，与生产者配置匹配。
组 ID 是强制性的，Kafka 使用组 ID 来允许并行数据消耗。ConcurrentKafkaListenerContainerFactoryBean 允许你的应用在多个线程中使用消息。
现在，您的 Java 应用已配置为在 Kafka 代理中查找使用者，让我们开始收听发送到主题的消息。
创建一个 src/main/java/com/okta/javakafka/consumer 目录，并在其中创建以下类：



Java