使用WSO2 Micro Integrator发送RabbitMQ消息报错求助

错误信息 #

The system cannot infer the transport information from the rabbitmq:/localhost:5672/rabbitqueue?rabbitmq.client.connection.retry.count=5&rabbitmq.message.content.type=application/json&rabbitmq.connection.username=guest&rabbitmq.connection.password=guest&rabbitmq.connection.virtual.host=/ URL.

已确认RabbitMQ服务正常运行，相关配置如下：

端点配置 #

<endpoint name="rabbitmq-endpoint">
  <address uri="rabbitmq:/localhost:5672/rabbitqueue?rabbitmq.client.connection.retry.count=5&amp;rabbitmq.message.content.type=application/json&amp;rabbitmq.connection.username=guest&amp;rabbitmq.connection.password=guest&amp;rabbitmq.connection.virtual.host=/"/>
</endpoint>

MessageStore配置 #

<?xml version="1.0" encoding="UTF-8"?>
<messageStore class="org.apache.synapse.message.store.impl.rabbitmq.RabbitMQStore" name="message-store" xmlns="http://ws.apache.org/ns/synapse">
    <parameter name="store.rabbitmq.host.name">localhost</parameter>
    <parameter name="store.producer.guaranteed.delivery.enable">false</parameter>
    <parameter name="store.rabbitmq.host.port">5672</parameter>
    <parameter name="store.rabbitmq.route.key"/>
    <parameter name="store.rabbitmq.username">guest</parameter>
    <parameter name="store.rabbitmq.virtual.host">/</parameter>
    <parameter name="rabbitmq.connection.ssl.enabled">false</parameter>
    <parameter name="store.rabbitmq.exchange.name"/>
    <parameter name="store.rabbitmq.queue.name">rabbitqueue</parameter>
    <parameter name="store.rabbitmq.password">guest</parameter>
</messageStore>

MessageProcessor配置 #

<?xml version="1.0" encoding="UTF-8"?>
<messageProcessor class="org.apache.synapse.message.processor.impl.forwarder.ScheduledMessageForwardingProcessor" messageStore="message-store" name="message-processor" targetEndpoint="rabbitmq-endpoint" xmlns="http://ws.apache.org/ns/synapse">
    <parameter name="client.retry.interval">1000</parameter>
    <parameter name="max.delivery.attempts">4</parameter>
    <parameter name="member.count">1</parameter>
    <parameter name="message.processor.reply.sequence">repseq</parameter>
    <parameter name="message.processor.fault.sequence">faultseq</parameter>
    <parameter name="store.connection.retry.interval">1000</parameter>
    <parameter name="max.store.connection.attempts">-1</parameter>
    <parameter name="max.delivery.drop">Disabled</parameter>
    <parameter name="interval">1000</parameter>
    <parameter name="is.active">true</parameter>
</messageProcessor>

完整堆栈日志 #

[2023-05-02 16:13:49,900] INFO {RabbitMQStore} - Store [message-store] Successfully connected to RabbitMQ Broker
[2023-05-02 16:13:49,916] ERROR {ClientUtils} - The system cannot infer the transport information from the rabbitmq:/localhost:5672/rabbitqueue?rabbitmq.client.connection.retry.count=5&rabbitmq.message.content.type=application/json&rabbitmq.connection.username=guest&rabbitmq.connection.password=guest&rabbitmq.connection.virtual.host=/ URL.
[2023-05-02 16:13:49,918] WARN {EndpointContext} - Endpoint : rrabbitmq:/localhost:5672/rabbitqueue?rabbitmq.client.connection.retry.count=5&rabbitmq.message.content.type=application/json&rabbitmq.connection.username=guest&rabbitmq.connection.password=guest&rabbitmq.connection.virtual.host=/ will be marked SUSPENDED as it failed
[2023-05-02 16:13:49,918] WARN {EndpointContext} - Suspending endpoint : rabbitmq-endpoint with address rabbitmq:/localhost:5672/rabbitqueue?rabbitmq.client.connection.retry.count=5&rabbitmq.message.content.type=application/json&rabbitmq.connection.username=guest&rabbitmq.connection.password=guest&rabbitmq.connection.virtual.host=/

• current suspend duration is : 30000ms - Next retry after : Tue May 02 16:14:19 IST 2023
  [2023-05-02 16:13:49,919] ERROR {ForwardingService} - [ message-processor ] Error while forwarding message to endpoint rabbitmq-endpoint.
  org.apache.synapse.SynapseException: Error sending Message to the endpoint
  at org.apache.synapse.message.processor.impl.forwarder.ForwardingService.tryToDispatchToEndpoint(ForwardingService.java:623)
  at org.apache.synapse.message.processor.impl.forwarder.ForwardingService.dispatch(ForwardingService.java:542)
  at org.apache.synapse.message.processor.impl.forwarder.ForwardingService.execute(ForwardingService.java:247)
  at org.wso2.micro.integrator.mediation.ntask.NTaskAdapter.execute(NTaskAdapter.java:105)
  at org.wso2.micro.integrator.ntask.core.impl.TaskQuartzJobAdapter.execute(TaskQuartzJobAdapter.java:63)
  at org.quartz.core.JobRunShell.run(JobRunShell.java:213)
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
  at java.util.concurrent.FutureTask.run(FutureTask.java:266)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  at java.lang.Thread.run(Thread.java:750)
  Caused by: org.apache.axis2.AxisFault: The system cannot infer the transport information from the rabbitmq:/localhost:5672/rabbitqueue?rabbitmq.client.connection.retry.count=5&rabbitmq.message.content.type=application/json&rabbitmq.connection.username=guest&rabbitmq.connection.password=guest&rabbitmq.connection.virtual.host=/ URL.
  at org.apache.axis2.description.ClientUtils.inferOutTransport(ClientUtils.java:86)
  at org.apache.axis2.client.OperationClient.prepareMessageContext(OperationClient.java:288)
  at org.apache.axis2.description.OutOnlyAxisOperationClient.executeImpl(OutOnlyAxisOperation.java:249)
  at org.apache.axis2.client.OperationClient.execute(OperationClient.java:149)
  at org.apache.synapse.message.senders.blocking.BlockingMsgSender.sendRobust(BlockingMsgSender.java:503)
  at org.apache.synapse.message.senders.blocking.BlockingMsgSender.send(BlockingMsgSender.java:380)
  at org.apache.synapse.core.axis2.Axis2FlexibleMEPClient.send(Axis2FlexibleMEPClient.java:102)
  at org.apache.synapse.core.axis2.Axis2Sender.sendOn(Axis2Sender.java:86)
  at org.apache.synapse.core.axis2.Axis2SynapseEnvironment.send(Axis2SynapseEnvironment.java:586)
  at org.apache.synapse.endpoints.AbstractEndpoint.send(AbstractEndpoint.java:409)
  at org.apache.synapse.endpoints.AddressEndpoint.send(AddressEndpoint.java:77)
  at org.apache.synapse.message.processor.impl.forwarder.ForwardingService.tryToDispatchToEndpoint(ForwardingService.java:619)
  ... 10 more
  [2023-05-02 16:13:49,921] ERROR {ForwardingService} - BlockingMessageSender of message processor [message-processor] failed to forward message to endpoint rabbitmq-endpoint
  [2023-05-02 16:13:49,922] ERROR {ForwardingService} - Message processor [message-processor] failed to send the message to client
  java.lang.NullPointerException
  at org.apache.synapse.message.processor.impl.forwarder.ForwardingService.sendThroughFaultSeq(ForwardingService.java:792)
  at org.apache.synapse.message.processor.impl.forwarder.ForwardingService.handleFailedInvocations(ForwardingService.java:690)
  at org.apache.synapse.message.processor.impl.forwarder.ForwardingService.tryToDispatchToEndpoint(ForwardingService.java:678)
  at org.apache.synapse.message.processor.impl.forwarder.ForwardingService.dispatch(ForwardingService.java:542)
  at org.apache.synapse.message.processor.impl.forwarder.ForwardingService.execute(ForwardingService.java:247)
  at org.wso2.micro.integrator.mediation.ntask.NTaskAdapter.execute(NTaskAdapter.java:105)
  at org.wso2.micro.integrator.ntask.core.impl.TaskQuartzJobAdapter.execute(TaskQuartzJobAdapter.java:63)
  at org.quartz.core.JobRunShell.run(JobRunShell.java:213)
  at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
  at java.util.concurrent.FutureTask.run(FutureTask.java:266)
  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
  at java.lang.Thread.run(Thread.java:750)

解决方案 #

1. 修正端点URI格式 #

RabbitMQ端点的URI需要使用双斜杠格式，并且指定交换器信息，修改后的端点配置如下：

<endpoint name="rabbitmq-endpoint">
  <address uri="rabbitmq://localhost:5672/rabbitqueue?rabbitmq.client.connection.retry.count=5&amp;rabbitmq.message.content.type=application/json&amp;rabbitmq.connection.username=guest&amp;rabbitmq.connection.password=guest&amp;rabbitmq.connection.virtual.host=/&amp;rabbitmq.exchange.name=amq.direct"/>
</endpoint>

• 将rabbitmq:/改为rabbitmq://（双斜杠是标准URI协议格式）
• 添加rabbitmq.exchange.name参数，使用默认direct交换器amq.direct，如果有自定义交换器可替换为对应名称

2. 检查RabbitMQ传输组件部署 #

确认WSO2 Micro Integrator的repository/components/plugins目录下存在以下RabbitMQ相关jar包：

• synapse-rabbitmq-transport_*.jar
• rabbitmq-client_*.jar
  若缺失，需补充对应版本的jar包后重启MI。

3. 确认Axis2传输配置 #

检查repository/conf/axis2/axis2.xml中是否添加了RabbitMQ传输配置，若没有则补充：

<transportSender name="rabbitmq" class="org.apache.synapse.transport.rabbitmq.RabbitMQSender"/>
<transportReceiver name="rabbitmq" class="org.apache.synapse.transport.rabbitmq.RabbitMQMessageReceiver"/>

添加后重启MI生效。

4. 优化MessageProcessor配置逻辑 #

当前配置中，MessageStore已经将消息存入RabbitMQ队列，再通过MessageProcessor转发到同RabbitMQ的端点属于重复操作：

• 如果需求是将消息存入RabbitMQ供外部消费，可直接移除MessageProcessor
• 如果需要转发到其他服务（如HTTP端点），修改targetEndpoint为对应服务的端点即可

5. 修复NullPointerException问题 #

堆栈中的空指针是因为转发失败后调用faultseq序列时出现异常，确保faultseq已正确定义并包含基础处理逻辑：

<sequence name="faultseq">
    <log level="full">
        <property name="ERROR_TYPE" value="MESSAGE_FORWARDING_FAILED"/>
    </log>
</sequence>

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