WSO2 ESB Kafka消费者项目如何通过IP端点实现消息自动消费

核心实现思路 #

你提到的IP端点就是WSO2 ESB的入站端点（Inbound Endpoint，常简称IP端点），官方原生提供了Kafka类型的入站端点，无需自定义底层IP层监听逻辑，开箱即可实现类似JMS消费者的主动监听消费能力：入站端点后台会常驻Kafka Consumer实例，持续监听指定Topic的新消息，收到消息后自动注入ESB的中介流程处理，完全符合你的需求。

具体配置步骤 #

1. 部署Kafka客户端依赖 #

• 下载和你的Kafka集群版本匹配的kafka-clients-x.x.x.jar依赖包，放到WSO2 ESB的/repository/components/lib目录下，重启ESB使依赖生效。

2. 创建Kafka入站端点 #

创建Kafka类型的入站端点，配置Kafka连接、消费相关参数，示例配置如下：

<inboundEndpoint name="KafkaConsumerInbound" protocol="kafka" sequence="KafkaMsgProcessSeq" onError="KafkaErrorHandleSeq" suspend="false">
    <parameters>
        <!-- Kafka集群地址，多节点用逗号分隔 -->
        <parameter name="bootstrap.servers">10.0.0.1:9092,10.0.0.2:9092</parameter>
        <!-- 待监听的目标Topic名称 -->
        <parameter name="topic.name">your_business_topic</parameter>
        <!-- 消费者组ID，同一组内消费者负载均衡消费 -->
        <parameter name="group.id">esb_consumer_group_01</parameter>
        <!-- 消息Key反序列化类 -->
        <parameter name="key.deserializer">org.apache.kafka.common.serialization.StringDeserializer</parameter>
        <!-- 消息Value反序列化类，可根据实际消息格式调整为ByteArray等其他反序列化类 -->
        <parameter name="value.deserializer">org.apache.kafka.common.serialization.StringDeserializer</parameter>
        <!-- 无偏移量记录时的消费起始位置，earliest=从最早未消费消息开始，latest=从最新消息开始 -->
        <parameter name="auto.offset.reset">earliest</parameter>
        <!-- 自动提交偏移量开关，关闭则需要在处理序列中手动提交偏移量 -->
        <parameter name="enable.auto.commit">true</parameter>
        <!-- 自动提交偏移量的间隔时间，单位毫秒 -->
        <parameter name="auto.commit.interval.ms">1000</parameter>
        <!-- 单次拉取最大消息数 -->
        <parameter name="max.poll.records">10</parameter>
        <!-- 消费线程数，建议和目标Topic的分区数保持一致，最大化消费能力 -->
        <parameter name="consumer.threads">3</parameter>
    </parameters>
</inboundEndpoint>

3. 编写业务处理序列 #

入站端点配置中指定的KafkaMsgProcessSeq为消息处理序列，Kafka消息拉取成功后会自动进入该序列执行你定义的业务逻辑，示例如下：

<sequence name="KafkaMsgProcessSeq" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
    <!-- 打印消费到的Kafka消息 -->
    <log level="full">
        <property name="ReceivedKafkaMsg" expression="$body"/>
    </log>
    <!-- 此处添加你的业务处理逻辑：报文转换、调用后端服务、写入存储等 -->
</sequence>

同时可配套编写KafkaErrorHandleSeq错误处理序列，消费出现异常时自动进入该序列，可实现重试、死信队列投递等容错逻辑。

兼容方案（适配无原生Kafka入站端点的旧版本ESB） #

如果你使用的ESB版本过老未提供Kafka类型的入站端点，可采用以下方案实现：

• 部署定时任务类型的入站端点，固定间隔触发自定义Class Mediator
• 在Class Mediator中自行实现Kafka Consumer的初始化、消息拉取、偏移量管理逻辑，拉取到消息后注入ESB中介流程处理
  该方案需要自行维护Kafka消费者的生命周期，维护成本相对较高，仅作为旧版本兼容使用。

内容的提问来源于stack exchange，提问作者ktraos