springmvc 集成rabbitmq，实现RPC通信方式

背景

最近在研究spring 集成rabbitmq 实现rpc通信，期间走了不少弯路。后来参考spring AMQP官方文档，终于实现了。RPC简单来说就是client发送mq消息，同步等待server端处理返回消息。

代码

1.首先gradle dependencies 引入 spring-amqp .jar。

compile 'org.springframework.amqp:spring-rabbit:1.6.6.RELEASE'

2.client 端配置文件<?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:rabbit="http://www.springframework.org/schema/rabbit"xmlns:context="http://www.springframework.org/schema/context"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:property-placeholder location="classpath*:/config/config.properties"/><!--连接rabbit server配置--><bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory"><property name="host" value="${rabbit.ip}"/><property name="port" value="${rabbit.port}"/><property name="username" value="${rabbit.username}"/><property name="password" value="${rabbit.password}"/></bean><!--对mq进行操作--><bean id="admin" class="org.springframework.amqp.rabbit.core.RabbitAdmin"><constructor-arg name="connectionFactory" ref="connectionFactory"/></bean><!--发送消息的队列--><rabbit:queue name="myQueue" durable="true"/><!--交换机，绑定队列--><rabbit:topic-exchange name="myExchange" durable="true"><rabbit:bindings><rabbit:binding queue="myQueue" pattern="foo.*"/></rabbit:bindings></rabbit:topic-exchange><!--rabbitmq模版--><bean id="amqpTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate"><constructor-arg ref="connectionFactory"/><property name="exchange" value="myExchange"/><property name="routingKey" value="foo.*"/><property name="queue" value="myQueue"/><!--接受响应超时时间--><property name="receiveTimeout" value="6000"/></bean></beans>3.controller 代码,发送mq消息。@Resourceprivate RabbitTemplate rabbitTemplate;@RequestMapping("/home")public String home(HttpServletRequest request) {try {Object o = rabbitTemplate.convertSendAndReceive("just a test");System.out.println(o);} catch (Exception e) {logger.error("-", e);}return "index";}4.server端配置文件<?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:rabbit="http://www.springframework.org/schema/rabbit"xmlns:context="http://www.springframework.org/schema/context"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <rabbit:connection-factory id="connectionFactory" port="5672" host="20.2.0.183"username="admin" password="admin"/><bean id="amqpTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate"><constructor-arg ref="connectionFactory"/><property name="exchange" value="myExchange"/><property name="queue" value="myQueue"/><property name="routingKey" value="foo.*"/><property name="replyQueue" ref="replyQ"/><property name="replyAddress" value="myExchange/foo.*"/><property name="replyTimeout" value="6000"/></bean><rabbit:admin connection-factory="connectionFactory" id="admin"/><!--处理myQueue发送的消息--><bean id="foo" class="com.*.*.*.MyListener"/><!--监听myQueue消息队列--><rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto"><rabbit:listener ref="foo" queue-names="myQueue"/></rabbit:listener-container><!--监听rabbitmq模版--><bean class="org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer"><constructor-arg ref="connectionFactory"/><property name="queues" ref="replyQ"/><property name="messageListener" ref="amqpTemplate"/><property name="acknowledgeMode" value="MANUAL"/></bean>

<!--应答myQueue的队列-->

<rabbit:queue id="replyQ" name="reply_to"/></beans>

5.MyListener 处理接受到mq发送的消息。

public class MyListener implements MessageListener {
@Resource
private RabbitTemplate rabbitTemplate;

@Override
public void onMessage(Message message) {
boolean result = rabbitTemplate.receiveAndReply("reply_to", new ReceiveAndReplyCallback<String, Map>() {
@Override
public Map handle(String s) {
System.out.println("receive:" + s);
Map map = new HashMap<String, String>();
map.put("reply", "success");
return map;
}
});
System.out.println("result :" + result);
}

}

运行 client 端代码，可以看到控制台打印的消息：client 打印的消息：

{reply=success}

server 打印的消息：

receive:just a test
result :true

以上代码就简单的实现了RPC通信。具体可以参考官方文档。