zoukankan      html  css  js  c++  java

  • RabbitMQ的原理和使用

    转载：RabbitMQ从入门到精通

    转载：轻松搞定RabbitMQ

    转载：RabbitMQ Java入门教程

    一、RabbitMQ

    AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。
    AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
    RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP等，支持AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

    二、RabbitMQ的使用场景

    对于一个大型的软件系统来说，它会有很多的组件或者说模块或者说子系统或者（subsystem or Component or submodule）。那么这些模块的如何通信？这和传统的IPC有很大的区别。传统的IPC很多都是在单一系统上的，模块耦合性很大，不适合扩展（Scalability）；如果使用socket那么不同的模块的确可以部署到不同的机器上，但是还是有很多问题需要解决。比如：
     1）信息的发送者和接收者如何维持这个连接，如果一方的连接中断，这期间的数据如何方式丢失？
     2）如何降低发送者和接收者的耦合度？
     3）如何让Priority高的接收者先接到数据？
     4）如何做到load balance？有效均衡接收者的负载？
     5）如何有效的将数据发送到相关的接收者？也就是说将接收者subscribe 不同的数据，如何做有效的filter。
     6）如何做到可扩展，甚至将这个通信模块发到cluster上？
     7）如何保证接收者接收到了完整，正确的数据？
      AMDQ协议解决了以上的问题，而RabbitMQ实现了AMQP。

    三、RabbitMQ的结构

    RabbitMQ的应用场景架构图如下：


    1. Broker：简单来说就是消息队列服务器实体。
    2. Exchange：消息交换机，它指定消息按什么规则，路由到哪个队列。
    3. Queue：消息队列载体，每个消息都会被投入到一个或多个队列。
    4. Binding：绑定，它的作用就是把exchange和queue按照路由规则绑定起来。
    5. Routing Key：路由关键字，exchange根据这个关键字进行消息投递。
    6. vhost：虚拟主机，一个broker里可以开设多个vhost，用作不同用户的权限分离。
    7. producer：消息生产者，就是投递消息的程序。
    8. consumer：消息消费者，就是接受消息的程序。
    9. channel：消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务。

    四、RabbitMQ的使用过程

    AMQP模型中，消息在producer中产生，发送到MQ的exchange上，exchange根据配置的路由方式发到相应的Queue上，Queue又将消息发送给consumer，消息从queue到consumer有push和pull两种方式。 消息队列的使用过程大概如下：

    1. 客户端连接到消息队列服务器，打开一个channel。
    2. 客户端声明一个exchange，并设置相关属性。
    3. 客户端声明一个queue，并设置相关属性。
    4. 客户端使用routing key，在exchange和queue之间建立好绑定关系。
    5. 客户端投递消息到exchange。
    exchange接收到消息后，就根据消息的key和已经设置的binding，进行消息路由，将消息投递到一个或多个队列里。 exchange也有几个类型，完全根据key进行投递的叫做Direct交换机，例如，绑定时设置了routing key为”abc”，那么客户端提交的消息，只有设置了key为”abc”的才会投递到队列。

    4.0 安装和配置

    RabbitMQ使用Erlang语言实现，因此在使用时首先要安装和配置erlang环境，并安装服务器后进行相关配置，由于不是本文主要内容所以忽略，详见RabbitMQ简介。

    RabbitMQ的客户端使用时需要添加相关依赖。

    4.1 点对点


    消息生产者的代码如下：

    1. package com.zenhobby.rabbit.demo;
    2.  
    3. import com.rabbitmq.client.Channel;
    4. import com.rabbitmq.client.Connection;
    5. import com.rabbitmq.client.ConnectionFactory;
    6.  
    7. public class Send
    8. {
    9. 	//队列名称
    10. 	private final static String QUEUE_NAME = "hello";
    11.  
    12. 	public static void main(String[] argv) throws java.io.IOException
    13. 	{
    14. 		/**
    15. 		 * 创建连接连接到MabbitMQ
    16. 		 */
    17. 		ConnectionFactory factory = new ConnectionFactory();
    18. 		//设置MabbitMQ所在主机ip或者主机名
    19. 		factory.setHost("localhost");
    20. 		//创建一个连接
    21. 		Connection connection = factory.newConnection();
    22. 		//创建一个频道
    23. 		Channel channel = connection.createChannel();
    24. 		//指定一个队列
    25. 		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    26. 		//发送的消息
    27. 		String message = "hello world!";
    28. 		//往队列中发出一条消息
    29. 		channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
    30. 		System.out.println(" [x] Sent '" + message + "'");
    31. 		//关闭频道和连接
    32. 		channel.close();
    33. 		connection.close();
    34. 	 }
    35. }
    消息消费者的代码如下： 

    1. package com.zenhobby.rabbit.demo;
    2.  
    3. import com.rabbitmq.client.Channel;
    4. import com.rabbitmq.client.Connection;
    5. import com.rabbitmq.client.ConnectionFactory;
    6. import com.rabbitmq.client.QueueingConsumer;
    7.  
    8. public class Recv
    9. {
    10. 	//队列名称
    11. 	private final static String QUEUE_NAME = "hello";
    12.  
    13. 	public static void main(String[] argv) throws java.io.IOException,
    14. 			java.lang.InterruptedException
    15. 	{
    16. 		//打开连接和创建频道，与发送端一样
    17. 		ConnectionFactory factory = new ConnectionFactory();
    18. 		factory.setHost("localhost");
    19. 		Connection connection = factory.newConnection();
    20. 		Channel channel = connection.createChannel();
    21. 		//声明队列，主要为了防止消息接收者先运行此程序，队列还不存在时创建队列。
    22. 		channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    23. 		System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    24. 		
    25. 		//创建队列消费者
    26. 		QueueingConsumer consumer = new QueueingConsumer(channel);
    27. 		//指定消费队列，关闭默认的消息应答
    28. 		channel.basicConsume(QUEUE_NAME, true, consumer);
    29. 		while (true)
    30. 		{
    31. 			//nextDelivery是一个阻塞方法（内部实现其实是阻塞队列的take方法）
    32. 			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    33. 			String message = new String(delivery.getBody());
    34. 			System.out.println(" [x] Received '" + message + "'");
    35. 		}
    36.  
    37. 	}
    38. }
    队列分别在生产者和消费者处创建，主要是为了防止有一端未建立起来的时候丢失消息。

    4.2 工作队列

    工作队列的主要任务是：避免立刻执行资源密集型任务，然后必须等待其完成。相反地，我们进行任务调度：我们把任务封装为消息发送给队列。工作进行在后台运行并不断的从队列中取出任务然后执行。当你运行了多个工作进程时，任务队列中的任务将会被工作进程共享执行。这样的概念在web应用中极其有用，当在很短的HTTP请求间需要执行复杂的任务。

    1.消息分发机制

    默认的，RabbitMQ会一个一个的发送信息给下一个消费者(consumer)，而不考虑每个任务的时长等等，且是一次性分配，并非一个一个分配。平均的每个消费者将会获得相等数量的消息。这样分发消息的方式叫做round-robin。

    默认的任务分发虽然看似公平但存在弊端。比如：现在有2个消费者，所有的奇数的消息都是繁忙的，而偶数则是轻松的。按照轮询的方式，奇数的任务交给了第一个消费者，所以一直在忙个不停。偶数的任务交给另一个消费者，则立即完成任务，然后闲得不行。而RabbitMQ则是不了解这些的。这是因为当消息进入队列，RabbitMQ就会分派消息。它不看消费者为应答的数目，只是盲目的将第n条消息发给第n个消费者。

    为了解决这个问题，我们使用basicQos( prefetchCount = 1)方法，来限制RabbitMQ只发不超过1条的消息给同一个消费者。当消息处理完毕后，有了反馈，才会进行第二次发送。

    1. int prefetchCount = 1;  
    2. channel.basicQos(prefetchCount);
    使用公平分发，必须关闭自动应答，改为手动应答。

    2. 消息确认

    每个Consumer可能需要一段时间才能处理完收到的数据。如果在这个过程中，Consumer出错了，异常退出了，而数据还没有处理完成，那么非常不幸，这段数据就丢失了。因为我们采用no-ack的方式进行确认，也就是说，每次Consumer接到数据后，而不管是否处理完成，RabbitMQ Server会立即把这个Message标记为完成，然后从queue中删除了。
    为了保证数据不被丢失，RabbitMQ支持消息确认机制，即acknowledgments。为了保证数据能被正确处理而不仅仅是被Consumer收到，那么我们不能采用no-ack。而应该是在处理完数据后发送ack。在处理数据后发送的ack，就是告诉RabbitMQ数据已经被接收，处理完成，RabbitMQ可以去安全的删除它了。如果Consumer退出了但是没有发送ack，那么RabbitMQ就会把这个Message发送到下一个Consumer。这样就保证了在Consumer异常退出的情况下数据也不会丢失。这里并没有用到超时机制。RabbitMQ仅仅通过Consumer的连接中断来确认该Message并没有被正确处理。也就是说，RabbitMQ给了Consumer足够长的时间来做数据处理。
    默认情况下，消息确认是打开的（enabled）：

    1. boolean autoAck = false;  
    2. channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    修改消费者如下： 

    1. channel.basicQos(1);//保证一次只分发一个  
    2. // 创建队列消费者  
    3. final Consumer consumer = new DefaultConsumer(channel) {  
    4.       @Override  
    5.       public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {  
    6.         String message = new String(body, "UTF-8");  
    7.   
    8.         System.out.println(" [x] Received '" + message + "'");  
    9.         System.out.println(" [x] Proccessing... at " +new Date().toLocaleString());  
    10.         try {  
    11.             for (char ch: message.toCharArray()) {  
    12.                 if (ch == '.') {  
    13.                     Thread.sleep(1000);  
    14.                 }  
    15.             }  
    16.         } catch (InterruptedException e) {  
    17.         } finally {  
    18.           System.out.println(" [x] Done! at " +new Date().toLocaleString());  
    19.           channel.basicAck(envelope.getDeliveryTag(), false);  
    20.         }  
    21.       }  
    22.     };
    其中： 

    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    用于在消息处理完毕时返回应答状态。如果MQ服务器未收到应答则在消费者挂掉之后重新把消息放入到队列中以供其他消费者使用。如果关闭了自动消息应答，手动也未设置应答，这是一个很简单的错误，但是后果却是极其严重的。消息在分发出去以后，得不到回应，所以不会在内存中删除，结果RabbitMQ会越来越占用内存，导致服务器挂掉。

    3. 消息持久化

    为了保证在RabbitMQ退出或者crash了数据仍没有丢失，需要将queue和Message都要持久化。

    queue的持久化需要在声明时指定durable=True：

    channel.queue_declare(queue='hello', durable=True)
    message的持久化需要在发送时指定property： 

    channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());

    修改后的生产者如下所示：

    1. static void Main(string[] args)
    2.         {
    3.             var factory = new ConnectionFactory() { HostName = "localhost" };
    4.             using (var connection = factory.CreateConnection())
    5.             {
    6.                 using (var channel = connection.CreateModel())
    7.                 {
    8.                     bool durable = true;
    9.                     channel.QueueDeclare("task_queue", durable, false, false, null);//queue的持久化需要在声明时指定durable=True
    10.                     var message = GetMessage(args);
    11.                     var body = Encoding.UTF8.GetBytes(message);
    12.                     var properties = channel.CreateBasicProperties();
    13.                     properties.SetPersistent(true);//需要持久化Message，即在Publish的时候指定一个properties，
    14.                     channel.BasicPublish("", "task_hello", properties, body);
    15.                 }
    16.             }
    17.         }

    4.3 Publish/Subscribe

    1. 交换器

    在工作队列一节中使用的分发如下：

    channel.basicPublish("", "hello", null, message.getBytes());
    其中第一个入参为空即为默认的交换器，交换器是RabbitMQ中的概念，其主要工作是接受生产者发出的消息，并推送到消息队列中（生产者并没有直接向queue中发送任何消息，而是发给交换器由交换器转交）。

    这里写图片描述
    交换器的规则有：

    1. direct （直连）：
    2. topic （主题）
    3. headers （标题）
    4. fanout （分发）

    Direct Exchange – 处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配。这是一个完整的匹配。如果一个队列绑定到该交换机上要求路由键 “dog”，则只有被标记为“dog”的消息才被转发，不会转发dog.puppy，也不会转发dog.guard，只会转发dog。 

    1. Channel channel = connection.createChannel();  
    2. channel.exchangeDeclare("exchangeName", "direct"); //direct fanout topic  
    3. channel.queueDeclare("queueName");  
    4. channel.queueBind("queueName", "exchangeName", "routingKey");  
    5.   
    6. byte[] messageBodyBytes = "hello world".getBytes();  
    7. //需要绑定路由键  
    8. channel.basicPublish("exchangeName", "routingKey", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);


    Fanout Exchange – 不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上。很像子网广播，每台子网内的主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。
    1. Channel channel = connection.createChannel();  
    2. channel.exchangeDeclare("exchangeName", "fanout"); //direct fanout topic  
    3. channel.queueDeclare("queueName");  
    4. channel.queueBind("queueName", "exchangeName", "routingKey");  
    5.   
    6. channel.queueDeclare("queueName1");  
    7. channel.queueBind("queueName1", "exchangeName", "routingKey1");  
    8.   
    9. byte[] messageBodyBytes = "hello world".getBytes();  
    10. //路由键需要设置为空  
    11. channel.basicPublish("exchangeName", "", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);


    Topic Exchange – 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词，符号“*”匹配不多不少一个词。因此“audit.#”能够匹配到“audit.irs.corporate”，但是“audit.*” 只会匹配到“audit.irs”。

    1. Channel channel = connection.createChannel();  
    2. channel.exchangeDeclare("exchangeName", "topic"); //direct fanout topic  
    3. channel.queueDeclare("queueName");  
    4. channel.queueBind("queueName", "exchangeName", "routingKey.*");  
    5.   
    6. byte[] messageBodyBytes = "hello world".getBytes();  
    7. channel.basicPublish("exchangeName", "routingKey.one", MessageProperties.PERSISTENT_TEXT_PLAIN, messageBodyBytes);

    Header Exchange

    Headers类型的exchange使用的比较少，它也是忽略routingKey的一种路由方式。是使用Headers来匹配的。

    Headers是一个键值对，可以定义成Hashtable。发送者在发送的时候定义一些键值对，接收者也可以再绑定时候传入一些键值对，两者匹配的话，则对应的队列就可以收到消息。匹配有两种方式all和any。这两种方式是在接收端必须要用键值"x-mactch"来定义。

    all代表定义的多个键值对都要满足，而any则代码只要满足一个就可以了。

    fanout，direct，topic exchange的routingKey都需要要字符串形式的，而headers exchange则没有这个要求，因为键值对的值可以是任何类型。

    消息生产者如下：

    1. package cn.slimsmart.rabbitmq.demo.headers;  
    2.   
    3. import java.util.Date;  
    4. import java.util.Hashtable;  
    5. import java.util.Map;  
    6.   
    7. import org.springframework.amqp.core.ExchangeTypes;  
    8.   
    9. import com.rabbitmq.client.AMQP;  
    10. import com.rabbitmq.client.AMQP.BasicProperties;  
    11. import com.rabbitmq.client.AMQP.BasicProperties.Builder;  
    12. import com.rabbitmq.client.Channel;  
    13. import com.rabbitmq.client.Connection;  
    14. import com.rabbitmq.client.ConnectionFactory;  
    15.   
    16. public class Producer {  
    17.     private final static String EXCHANGE_NAME = "header-exchange";  
    18.       
    19.     @SuppressWarnings("deprecation")  
    20.     public static void main(String[] args) throws Exception {  
    21.         // 创建连接和频道  
    22.         ConnectionFactory factory = new ConnectionFactory();  
    23.         factory.setHost("192.168.36.102");  
    24.         // 指定用户 密码  
    25.         factory.setUsername("admin");  
    26.         factory.setPassword("admin");  
    27.         // 指定端口  
    28.         factory.setPort(AMQP.PROTOCOL.PORT);  
    29.         Connection connection = factory.newConnection();  
    30.         Channel channel = connection.createChannel();  
    31.           
    32.         //声明转发器和类型headers  
    33.         channel.exchangeDeclare(EXCHANGE_NAME, ExchangeTypes.HEADERS,false,true,null);  
    34.         String message = new Date().toLocaleString() + " : log something";  
    35.           
    36.         Map<String,Object> headers =  new Hashtable<String, Object>();  
    37.         headers.put("aaa", "01234");  
    38.         Builder properties = new BasicProperties.Builder();  
    39.         properties.headers(headers);  
    40.           
    41.         // 指定消息发送到的转发器,绑定键值对headers键值对  
    42.         channel.basicPublish(EXCHANGE_NAME, "",properties.build(),message.getBytes());  
    43.           
    44.         System.out.println("Sent message :'" + message + "'");  
    45.         channel.close();  
    46.         connection.close();  
    47.     }  
    48. }

    消息消费者如下：

    1. package cn.slimsmart.rabbitmq.demo.headers;  
    2.   
    3. import java.util.Hashtable;  
    4. import java.util.Map;  
    5.   
    6. import org.springframework.amqp.core.ExchangeTypes;  
    7.   
    8. import com.rabbitmq.client.AMQP;  
    9. import com.rabbitmq.client.Channel;  
    10. import com.rabbitmq.client.Connection;  
    11. import com.rabbitmq.client.ConnectionFactory;  
    12. import com.rabbitmq.client.QueueingConsumer;  
    13.   
    14. public class Consumer {  
    15.     private final static String EXCHANGE_NAME = "header-exchange";  
    16.     private final static String QUEUE_NAME = "header-queue";  
    17.       
    18.     public static void main(String[] args) throws Exception {  
    19.         // 创建连接和频道  
    20.         ConnectionFactory factory = new ConnectionFactory();  
    21.         factory.setHost("192.168.36.102");  
    22.         // 指定用户 密码  
    23.         factory.setUsername("admin");  
    24.         factory.setPassword("admin");  
    25.         // 指定端口  
    26.         factory.setPort(AMQP.PROTOCOL.PORT);  
    27.         Connection connection = factory.newConnection();  
    28.         Channel channel = connection.createChannel();  
    29.           
    30.         //声明转发器和类型headers  
    31.         channel.exchangeDeclare(EXCHANGE_NAME, ExchangeTypes.HEADERS,false,true,null);  
    32.         channel.queueDeclare(QUEUE_NAME,false, false, true,null);  
    33.           
    34.         Map<String, Object> headers = new Hashtable<String, Object>();  
    35.         headers.put("x-match", "any");//all any  
    36.         headers.put("aaa", "01234");  
    37.         headers.put("bbb", "56789");  
    38.         // 为转发器指定队列，设置binding 绑定header键值对  
    39.         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME,"", headers);  
    40.         QueueingConsumer consumer = new QueueingConsumer(channel);  
    41.         // 指定接收者，第二个参数为自动应答，无需手动应答  
    42.         channel.basicConsume(QUEUE_NAME, true, consumer);  
    43.         while (true) {  
    44.             QueueingConsumer.Delivery delivery = consumer.nextDelivery();  
    45.             String message = new String(delivery.getBody());  
    46.             System.out.println(message);  
    47.         }   
    48.     }  
    49. }


    Default Exchange

    其实除了上面四种以外还有一种Default Exchange，它是一种特别的Direct Exchange。
    当你手动创建一个队列时，后台会自动将这个队列绑定到一个名称为空的Direct类型交换机上，绑定路由名称与队列名称相同。有了这个默认的交换机和绑定，我们就可以像其他轻量级的队列，如Redis那样，直接操作队列来处理消息。不过只是看起来是，实际上在RabbitMQ里直接操作是不可能的。消息始终都是先发送到交换机，由交换级经过路由传送给队列，消费者再从队列中获取消息的。不过由于这个默认交换机和路由的关系，使我们只关心队列这一层即可，这个比较适合做一些简单的应用，毕竟没有发挥RabbitMQ的最大功能，如果都用这种方式去使用的话就真是杀鸡用宰牛刀了。

    2. 临时队列

    如果要在生产者和消费者之间创建一个新的队列，又不想使用原来的队列，临时队列就是为这个场景而生的：
    首先，每当我们连接到RabbitMQ，我们需要一个新的空队列，我们可以用一个随机名称来创建，或者说让服务器选择一个随机队列名称给我们。
    一旦我们断开消费者，队列应该立即被删除。Java客户端提供queuedeclare()为我们创建一个非持久化、独立、自动删除的队列名称。

    String queueName = channel.queueDeclare().getQueue();
    通过上面的代码就能获取到一个随机队列名称。 例如：它可能是：amq.gen-jzty20brgko-hjmujj0wlg。

    3. 绑定

    这里写图片描述

    如果我们已经创建了一个分发交换器和队列，现在我们就可以就将我们的队列跟交换器进行绑定。

    channel.queueBind(queueName, "logs", "");
    执行完这段代码后，日志交换器会将消息添加到我们的队列中。


    五、RabbitMQ实现RPC

    RabbitMQ可以用于实现RPC，两者有相像之处，使用RabbitMQ实现RPC分为如下几个步骤：

    1. Client interface（客户端接口）

    为了说明RPC服务可以使用，我们创建一个简单的客户端类。暴露一个方法——发送RPC请求，然后阻塞直到获得结果。
     

    1. FibonacciRpcClient fibonacciRpc = new FibonacciRpcClient();     
    2. String result = fibonacciRpc.call("4");  
    3. System.out.println( "fib(4) is " + result);

    2. Callback queue（回调队列）

    一般在RabbitMQ中做RPC是很简单的。客户端发送请求消息，服务器回复响应的消息。为了接受响应的消息，我们需要在请求消息中发送一个回调队列。可以用默认的队列：

    1. BasicProperties props = new BasicProperties  
    2.                             .Builder()  
    3.                             .replyTo(callbackQueueName)  
    4.                             .build();  
    5.   
    6. channel.basicPublish("", "rpc_queue", props, message.getBytes());  
    7.   
    8. // ... then code to read a response message from the callback_queue ...

    3. Message properties（消息属性）

    AMQP协议为消息预定义了一组14个属性。大部分的属性是很少使用的。除了一下几种：

    1. deliveryMode：标记消息传递模式，2-消息持久化，其他值-瞬态。在第二篇文章中还提到过。
    2. contentType：内容类型，用于描述编码的mime-type。例如经常为该属性设置JSON编码。
    3. replyTo：应答，通用的回调队列名称
    4. correlationId：关联ID，方便RPC响应与请求关联
    我们需要添加一个新的导入： 

    import com.rabbitmq.client.AMQP.BasicProperties;

    4. Correlation Id

    在上述方法中为每个RPC请求创建一个回调队列。这是很低效的。幸运的是，一个解决方案：可以为每个客户端创建一个单一的回调队列。
    新的问题被提出，队列收到一条回复消息，但是不清楚是那条请求的回复。这是就需要使用correlationId属性了。我们要为每个请求设置唯一的值。然后，在回调队列中获取消息，看看这个属性，关联response和request就是基于这个属性值的。如果我们看到一个未知的correlationId属性值的消息，可以放心的无视它——它不是我们发送的请求。
    你可能问道，为什么要忽略回调队列中未知的信息，而不是当作一个失败？这是由于在服务器端竞争条件的导致的。虽然不太可能，但是如果RPC服务器在发送给我们结果后，发送请求反馈前就挂掉了，这有可能会发送未知correlationId属性值的消息。如果发生了这种情况，重启RPC服务器将会重新处理该请求。这就是为什么在客户端必须很好的处理重复响应，RPC应该是幂等的。

    5. 实现


    我们的RPC的处理流程：

    1. 当客户端启动时，创建一个匿名的回调队列。
    2. 客户端为RPC请求设置2个属性：replyTo：设置回调队列名字；correlationId：标记request。
    3. 请求被发送到rpc_queue队列中。
    4. RPC服务器端监听rpc_queue队列中的请求，当请求到来时，服务器端会处理并且把带有结果的消息发送给客户端。接收的队列就是replyTo设定的回调队列。
    5. 客户端监听回调队列，当有消息时，检查correlationId属性，如果与request中匹配，那就是结果了。

    RPC服务器端（RPCServer.java）

    1. /**
    2.  * RPC服务器端
    3.  * 
    4.  * @author arron
    5.  * @date 2015年9月30日 下午3:49:01
    6.  * @version 1.0
    7.  */
    8. public class RPCServer {
    9. 	
    10. 	private static final String RPC_QUEUE_NAME = "rpc_queue";
    11.  
    12. 	public static void main(String[] args) throws Exception {
    13. 		ConnectionFactory factory = new ConnectionFactory();
    14. 		// 设置MabbitMQ所在主机ip或者主机名
    15. 		factory.setHost("127.0.0.1");
    16. 		// 创建一个连接
    17. 		Connection connection = factory.newConnection();
    18. 		// 创建一个频道
    19. 		Channel channel = connection.createChannel();
    20.  
    21. 		//声明队列
    22. 		channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
    23.  
    24. 		//限制：每次最多给一个消费者发送1条消息
    25. 		channel.basicQos(1);
    26.  
    27. 		//为rpc_queue队列创建消费者，用于处理请求
    28. 		QueueingConsumer consumer = new QueueingConsumer(channel);
    29. 		channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
    30.  
    31. 		System.out.println(" [x] Awaiting RPC requests");
    32.  
    33. 		while (true) {
    34. 			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    35.  
    36. 			//获取请求中的correlationId属性值，并将其设置到结果消息的correlationId属性中
    37. 			BasicProperties props = delivery.getProperties();
    38. 			BasicProperties replyProps = new BasicProperties.Builder().correlationId(props.getCorrelationId()).build();
    39. 			//获取回调队列名字
    40. 			String callQueueName = props.getReplyTo();
    41. 			
    42. 			String message = new String(delivery.getBody(),"UTF-8");
    43. 			
    44. 			System.out.println(" [.] fib(" + message + ")");
    45. 			
    46. 			//获取结果
    47. 			String response = "" + fib(Integer.parseInt(message));
    48. 			//先发送回调结果
    49. 			channel.basicPublish("", callQueueName, replyProps,response.getBytes());
    50. 			//后手动发送消息反馈
    51. 			channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
    52. 		}
    53. 	}
    54.  
    55. 	/**
    56. 	 * 计算斐波列其数列的第n项
    57. 	 * 
    58. 	 * @param n
    59. 	 * @return
    60. 	 * @throws Exception
    61. 	 */
    62. 	private static int fib(int n) throws Exception {
    63. 		if (n < 0)
    64. 			throw new Exception("参数错误，n必须大于等于0");
    65. 		if (n == 0)
    66. 			return 0;
    67. 		if (n == 1)
    68. 			return 1;
    69. 		return fib(n - 1) + fib(n - 2);
    70. 	}
    71. }
    RPC客户端（RPCClient.java）： 

    1. /** 
    2.  * 
    3.  * @author arron
    4.  * @date 2015年9月30日 下午3:44:43 
    5.  * @version 1.0 
    6.  */
    7. public class RPCClient {
    8.  
    9. 	private static final String RPC_QUEUE_NAME = "rpc_queue";
    10. 	
    11. 	private Connection connection;
    12. 	private Channel channel;
    13. 	private String replyQueueName;
    14. 	private QueueingConsumer consumer;
    15. 	
    16. 	public RPCClient() throws Exception {
    17. 	    ConnectionFactory factory = new ConnectionFactory();
    18. 		// 设置MabbitMQ所在主机ip或者主机名
    19. 		factory.setHost("127.0.0.1");
    20. 		// 创建一个连接
    21. 		connection = factory.newConnection();
    22. 		// 创建一个频道
    23. 		channel = connection.createChannel();
    24. 		
    25. 		//声明队列
    26. 		channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
    27. 		
    28. 		//为每一个客户端获取一个随机的回调队列
    29. 	    replyQueueName = channel.queueDeclare().getQueue();
    30. 	    //为每一个客户端创建一个消费者（用于监听回调队列，获取结果）
    31. 	    consumer = new QueueingConsumer(channel);
    32. 	    //消费者与队列关联
    33. 	    channel.basicConsume(replyQueueName, true, consumer);
    34. 	}
    35. 	
    36. 	/**
    37. 	 * 获取斐波列其数列的值
    38. 	 * 
    39. 	 * @param message
    40. 	 * @return
    41. 	 * @throws Exception
    42. 	 */
    43. 	public String call(String message) throws Exception{
    44. 		String response = null;
    45. 	    String corrId = java.util.UUID.randomUUID().toString();
    46.  
    47. 	    //设置replyTo和correlationId属性值
    48. 	    BasicProperties props = new BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();
    49.  
    50. 	    //发送消息到rpc_queue队列
    51. 	    channel.basicPublish("", RPC_QUEUE_NAME, props, message.getBytes());
    52.  
    53. 	    while (true) {
    54. 	        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    55. 	        if (delivery.getProperties().getCorrelationId().equals(corrId)) {
    56. 	            response = new String(delivery.getBody(),"UTF-8");
    57. 	            break;
    58. 	        }
    59. 	    }
    60.  
    61. 	    return response; 
    62. 	}
    63.  
    64. 	public static void main(String[] args) throws Exception {
    65. 		RPCClient fibonacciRpc = new RPCClient();   
    66. 		String result = fibonacciRpc.call("4");
    67. 		System.out.println( "fib(4) is " + result);
    68. 	}
    69. }

    这里的例子只是RabbitMQ中RPC服务的一个实现，你也可以根据业务需要实现更多。rpc有一个优点，如果一个RPC服务器处理不来，可以再增加一个、两个、三个。我们的例子中的代码还比较简单，还有很多问题没有解决：
    • 如果没有发现服务器，客户端如何处理？
    • 如果客户端的RPC请求超时了怎么办？
    • 如果服务器出现了故障，发生了异常，是否将异常发送到客户端？
    • 在处理消息前，怎样防止无效的消息？检查范围、类型？

    以上内容转载自网络，详见开头转载声明。
  • 相关阅读:
    高性能反射初体验2
    高性能反射初体验1
    HTML解析原理
    QQ邮箱漂流瓶的"变化"~~
    javascript跟随滚动条滚动,onscroll事件的学习
    认识一下window.location.hash
    IE中替换a标签href属性的一个bug
    用css Sprites的时候IE6的一个小bug
    css中如何引入“非常” 字体
    [重温经典]ie6 不支持position:fix的解决方案
  • 原文地址：https://www.cnblogs.com/jpfss/p/10313131.html
Copyright © 2011-2022 走看看