spring对rabbitmq RPC的支持

spring 整合 rabbitmq 的项目为 spring-amqp.  spring rabbitmq 支持RPC。需要使用spring-rabbit-support.

<!-- https://mvnrepository.com/artifact/org.springframework.amqp/spring-rabbit --> <dependency> <groupId>org.springframework.amqp</groupId> <artifactId>spring-rabbit</artifactId> <version>1.7.3.RELEASE</version> </dependency> 1234567 1234567

spring 通过 rabbitmq 支持RPC，采用Java序列化和AMQP协议。

接口源代码：

package net.oschina.rpc; public interface Service { int add(int x, int y); } 1234567 1234567

实现源代码：

package net.oschina.rpc; public class ServiceImp implements Service { public int add(int x, int y) { return x+y; } } 1234567891011 1234567891011

创建客户端配置文件

<?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" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.7.xsd"> <!--注入服务接口--> <bean id="client" class="org.springframework.amqp.remoting.client.AmqpProxyFactoryBean"> <property name="amqpTemplate" ref="template" /> <property name="serviceInterface" value="net.oschina.rpc.Service" /> </bean> <!--rabbit连接--> <rabbit:connection-factory id="connectionFactory" host="ip" username="root" password="password" /> <rabbit:admin connection-factory="connectionFactory" /> <!--创建rabbit 模板--> <rabbit:template id="template" connection-factory="connectionFactory" reply-timeout="2000" routing-key="remoting.binding" exchange="remoting.exchange" /> <rabbit:admin connection-factory="connectionFactory" /> <!--用于缓存remoting请求的队列--> <rabbit:queue name="remoting.queue" /> <rabbit:direct-exchange name="remoting.exchange"> <rabbit:bindings> <rabbit:binding queue="remoting.queue" key="remoting.binding" /> </rabbit:bindings> </rabbit:direct-exchange> </beans> 1234567891011121314151617181920212223242526272829303132 1234567891011121314151617181920212223242526272829303132

创建服务器端配置文件

<?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" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.0.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit-1.7.xsd"> <!--服务端封装--> <bean id="listener" class="org.springframework.amqp.remoting.service.AmqpInvokerServiceExporter"> <!--服务接口--> <property name="serviceInterface" value="net.oschina.rpc.Service" /> <property name="service" ref="service" /> <property name="amqpTemplate" ref="template" /> </bean> <!--接口实现的封装--> <bean id="service" class="net.oschina.rpc.ServiceImp" /> <rabbit:connection-factory id="connectionFactory" host="ip" username="root" password="password" /> <rabbit:admin connection-factory="connectionFactory" /> <rabbit:template id="template" connection-factory="connectionFactory" /> <rabbit:queue name="remoting.queue" /> <!--rabbit服务监听 prefetch 设置为1 为每次获取一个请求--> <!--消费者接收消息超时时长，receive-timeout 单位为毫秒--> <rabbit:listener-container prefetch="1" receive-timeout="2000" connection-factory="connectionFactory"> <rabbit:listener ref="listener" queue-names="remoting.queue" /> </rabbit:listener-container> </beans> 1234567891011121314151617181920212223242526272829303132 1234567891011121314151617181920212223242526272829303132

服务器端代码：

import org.springframework.context.support.AbstractApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; public class App { public static void main(String[] args) throws InterruptedException { AbstractApplicationContext ctx = new ClassPathXmlApplicationContext("rabbit-server.xml"); } } 123456789101112131415 123456789101112131415

客户端调用代码：

import org.springframework.context.support.ClassPathXmlApplicationContext; import net.oschina.rpc.Service; public class App2 { private static ClassPathXmlApplicationContext ctx; public static void main(String[] args) { ctx = new ClassPathXmlApplicationContext("client.xml"); Service service = ctx.getBean(Service.class); int i = service.add(7, 23); System.out.println(i); ctx.destroy(); } } 123456789101112131415161718 123456789101112131415161718

---

服务端日志

EBUG] [2017-06-09 15:38:45] org.springframework.amqp.rabbit.listener.BlockingQueueConsumer$InternalConsumer.handleDelivery(829) | Storing delivery for Consumer@6b419da: tags=[{amq.ctag-Pz6I0WhtxbKgR5li4-HoMw=remoting.queue}], channel=Cached Rabbit Channel: AMQChannel(amqp://root@120.92.73.218:5672/,1), conn: Proxy@b24502d Shared Rabbit Connection: SimpleConnection@666138da [delegate=amqp://root@120.92.73.218:5672/, localPort= 62043], acknowledgeMode=AUTO local queue size=0  [DEBUG] [2017-06-09 15:38:46] org.springframework.amqp.rabbit.listener.BlockingQueueConsumer.handle(418) | Received message: (Body:’RemoteInvocation: method name ‘add’; parameter types [int, int]’ MessageProperties [headers={}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=amq.rabbitmq.reply-to.g2dkABVyYWJiaXRAdm0xNzItMzEtMTYtMTIAAAS4AAAAAAI=.+iFiCAvwbRdzkBP01tN2Tw==, contentType=application/x-java-serialized-object, contentEncoding=null, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=remoting.exchange, receivedRoutingKey=remoting.binding, receivedDelay=null, deliveryTag=1, messageCount=0, consumerTag=amq.ctag-Pz6I0WhtxbKgR5li4-HoMw, consumerQueue=remoting.queue])  [DEBUG] [2017-06-09 15:38:46] org.springframework.amqp.rabbit.connection.CachingConnectionFactory.getCachedChannelProxy(494) | Creating cached Rabbit Channel from AMQChannel(amqp://root@120.92.73.218:5672/,2)  [DEBUG] [2017-06-09 15:38:46] org.springframework.amqp.rabbit.core.RabbitTemplate.doExecute(1453) | Executing callback on RabbitMQ Channel: Cached Rabbit Channel: AMQChannel(amqp://root@120.92.73.218:5672/,2), conn: Proxy@b24502d Shared Rabbit Connection: SimpleConnection@666138da [delegate=amqp://root@120.92.73.218:5672/, localPort= 62043]  [DEBUG] [2017-06-09 15:38:46] org.springframework.amqp.rabbit.core.RabbitTemplate.doSend(1507) | Publishing message on exchange [], routingKey = [amq.rabbitmq.reply-to.g2dkABVyYWJiaXRAdm0xNzItMzEtMTYtMTIAAAS4AAAAAAI=.+iFiCAvwbRdzkBP01tN2Tw==]

从日志可以看到，服务端从remoting.binding这个队列获取RPC请求信息。完成处理后反馈到一个特定路由键，客户端通过这个路由键获取处理后的结果。

客户端日志：

[DEBUG] [2017-06-09 15:38:45] org.springframework.amqp.rabbit.core.RabbitTemplate.doSend(1507) | Publishing message on exchange [remoting.exchange], routingKey = [remoting.binding]  [DEBUG] [2017-06-09 15:38:46] org.springframework.amqp.rabbit.core.RabbitTemplate.exchangeMessages(1380) | Reply: (Body:’org.springframework.remoting.support.RemoteInvocationResult@6d763516’ MessageProperties [headers={}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=application/x-java-serialized-object, contentEncoding=null, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=, receivedRoutingKey=amq.rabbitmq.reply-to.g2dkABVyYWJiaXRAdm0xNzItMzEtMTYtMTIAAAS4AAAAAAI=.+iFiCAvwbRdzkBP01tN2Tw==, receivedDelay=null, deliveryTag=1, messageCount=null, consumerTag=null, consumerQueue=null])