Spring 集成 - Apache ActiveMQ 到 Kafka

debugcn 发表于 Dev

瓦斯基塔库尔

我正在使用以下配置将 activemq 与 kafka 集成。我收到来自 activemq 的消息并将其转发给 kafka。但是，我注意到消息从 JMS 队列中出队，但消息不会进入 kafka。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:jms="http://www.springframework.org/schema/integration/jms"
    xmlns:integration="http://www.springframework.org/schema/integration"
    xmlns:int-kafka="http://www.springframework.org/schema/integration/kafka"
    xmlns:task="http://www.springframework.org/schema/task"
    xsi:schemaLocation="http://www.springframework.org/schema/beans 
    http://www.springframework.org/schema/beans/spring-beans.xsd 
    http://www.springframework.org/schema/integration/jms 
    http://www.springframework.org/schema/integration/jms/spring-integration-jms.xsd
    http://www.springframework.org/schema/integration/kafka 
    http://www.springframework.org/schema/integration/kafka/spring-integration-kafka.xsd">

    <jms:message-driven-channel-adapter
        id="helloJMSAdapater" destination="helloJMSQueue" connection-factory="jmsConnectionfactory"
        channel="helloChannel" extract-payload="true" />

    <integration:channel id="helloChannel" />

    <integration:service-activator id="sayHelloServiceActivator"
        input-channel="helloChannel" ref="sayHelloService" method="sayHello" />

    <int-kafka:outbound-channel-adapter
        id="kafkaOutboundChannelAdapter" kafka-template="template"
        auto-startup="false" sync="true" channel="helloChannel" topic="test1234" 
        >
    </int-kafka:outbound-channel-adapter>

    <bean id="template" class="org.springframework.kafka.core.KafkaTemplate">
        <constructor-arg>
            <bean class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
                <constructor-arg>
                    <map>
                        <entry key="bootstrap.servers" value="localhost:9092" />
                        <!--entry key="retries" value="5" /> <entry key="batch.size" value="16384" 
                            /> <entry key="linger.ms" value="1" /> <entry key="buffer.memory" value="33554432" 
                            /> < entry key="key.serializer" value="org.apache.kafka.common.serialization.StringSerializer" 
                            /> <entry key="value.serializer" value="org.apache.kafka.common.serialization.StringSerializer" 
                            / -->
                    </map>
                </constructor-arg>
            </bean>
        </constructor-arg>
    </bean>



</beans>

此外，如果 Kafka 出现任何问题，它甚至不会报告任何异常堆栈跟踪。

我错过了什么吗？

哈森·本努尔

您的消息由 sayHelloServiceActivator 使用。

因此，将您的 helloChannel 频道类型更改为

<publish-subscribe-channel id="helloChannel"/>

默认为 DirectChannel

DirectChannel 具有点对点语义，但除此之外，与上述任何基于队列的通道实现相比，它更类似于 PublishSubscribeChannel。它实现了 SubscribableChannel 接口而不是 PollableChannel 接口，因此它将消息直接分派给订阅者。但是，作为点对点通道，它与 PublishSubscribeChannel 的不同之处在于它只会将每个 Message 发送到单个订阅的 MessageHandler。

本文收集自互联网，转载请注明来源。

如有侵权，请联系[email protected] 删除。