rabbitmq用于分布式系统

上文介绍了RabbitMQ在linux下的安装，这里就简单的介绍一下基于RabbitMQ的开发。RabbitMQ已经提供了一大坨材料。

Java –http://www.RabbitMQ.com/java-client.html
.NET –http://www.RabbitMQ.com/dotnet.html
Erlang–http://www.RabbitMQ.com/erlang-client.html

这里使用的RabbitMQ提供的Java的包。

RabbitMQ提供的Java AMQP Client Libaray，最核心的class是ConnectionFactory、Conncetion 和Channel，分别代表连结工厂、一个连结和一个数据通道。

AMQP支持在一个TCP连接上启用多个MQ通信channel，每个channel都可以被应用作为通信流。每个AMQP程序至少要有一个connection和一个channel。

分布式的消息服务器肯定是要有Producer和Comsumer的。那怎么基于RabbitMQ构建一个Comsumer呢？

ConnectionFactory factory = new ConnectionFactory();
factory.setUsername(userName);
factory.setPassword(password);
factory.setVirtualHost(virtualHost);
factory.setHost(hostName);
factory.setPort(portNumber);
Connection conn = factory.newConnection();

ConnectionFactory factory = new ConnectionFactory();
factory.setUri("amqp://userName:password@hostName:portNumber/virtualHost");
Connection conn = factory.newConnection();

上述的代码，创建了一个connnection，开发者可以在一个connnection创建多个channel。每个channel都被分配了一个整数标识，自动由Connection类的createChannel()方法维护。或者，你可以使用.createChannel(x)来指定channel标识，其中x是你想要使用的channel标识。通常情况下，推荐使用.channel()方法来自动分配channel标识，以便防止冲突。channel用于传递、接收消息。如下代码：

Channel channel = conn.createChannel();

另外，获取connection时候，consumer线程默认自动分配一个 ExceutorService thread pool，如果你需要自己控制线程池的。你可以如下定义：

ExecutorService es = Executors.newFixedThreadPool(20);
Connection conn = factory.newConnection(es);

但是需要注意的是：当connection shutdown的时候,默认分配的ExecutorService会自动shutdown，用户自己定义的不会shutdown，需要手动shutdown。而且，只用在Consumer回调有严重的性能瓶颈的时候才用这种方式，在平常的使用中，默认的分配更加高效。

现在，我们已经有了一个可用的连结和channel,我们来创建一个队列。

//queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,Map<String, Object> arguments) throws IOException;
channel.queueDeclare(”vinoQueue",true,false,false,null);

它创建了一个名叫“vinoQueue”的队列，它是non-durable的（重启之后不会重新建立），并且最后一个消费者断开的时候不会自动删除（autoDelete=False）。在创建durable的队列（或者交换机）的时候，将auto_delete设置成false是很重要的，否则队列将会在最后一个消费者断开的时候消失。另外，还有一个属性是”exclusive“，该属性如果设置成True，只有创建这个队列的消费者程序才允许连接到该队列，这种队列对于这个消费者程序是私有的。

现在队列已经创建成功了，Comsumer有两种方式从队列中取出消息。

一个是调用channel.basicGet(“vinoQueue”, false)主动从队列中拉出下一个消息。如果队列中无消息，会返回null，代码如下所示：

// GetResponse basicGet(String queue, boolean autoAck) throws IOException;  主动从队列里拉出下一条消息。
channel.basicGet(QUEUE_NAME, false);

另一个是消息队列把消息推给Cumsumer。这种情况你就需要用如下代码。

//消费者用于获取消息信道绑定的消息队列中的信息
QueueingConsumer consumer = new QueueingConsumer(channel);
//注册一个新消息到达的回调，第二个参数表示ack，默认是false.
channel.basicConsume(QUEUE_NAME, true,consumer);

while(true){
//循环获取消息队列中的信息，这个函数会等待在队列上，直到下一个消息到达队列。
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
System.out.println("[x] Received '"+message+"'");

}
/用来注销该回调函数,参数consumer_tag 当中指定的字符串和chan.basicConsume() 注册的一致，这里无用。
//channel.basicCancel(QUEUE_NAME);

这里具体的说一下channel.basicConsume()里的ack参数。当从队列当中取出一个消息的时候，RabbitMQ需要应用显式地回馈说已经获取到了该消息。如果一段时间内不回馈，RabbitMQ会将该消息重新分配给另外一个绑定在该队列上的消费者。另一种情况是消费者断开连接，但是获取到的消息没有回馈，则RabbitMQ同样重新分配。如果将该参数设置为true，则RabbimtMQ会为下一个AMQP请求添加一个ack属性，告诉AMQP服务器需要等待回馈。否者，不要等待回馈。大多数时候，你也许想要自己手工发送回馈，例如，需要在回馈之前将消息存入数据库。回馈通常是通过调用channel.basicAck(deliveryTag, multiple)方法。

以上就是Comsumer的相关代码。

不过没有人发送消息的话，要消费者何用？所以需要一个Productor。消息投递的大概过程是：

客户端连接到消息队列，创建一个channel.
声明exchange
声明queue
绑定exchange和queue
投递消息到exchange

下面代码表示如果将一个简单的消息发送到名为“vino”的exchange，并且标记为路由键“hello”

// 创建消息通道
        Channel channel = connection.createChannel();

        /**
         * 声明Exchange 
         * Exchange.DeclareOk exchangeDeclare(String exchange, String
         * type, boolean durable, boolean autoDelete, Map<String, Object>
         * argumen s) throws IOException;
         */
        channel.exchangeDeclare("vino", "direct", true, false, null);
        /**
         * Declaring a queue is idempotent - it will only be created if it
         * doesn't exist already. The message content is a byte array, so you
         * can encode whatever you like there.
         */
        channel.queueDeclare("vinoQueue", false, false, false, null);
        // 绑定exchange和消息队列 Queue.BindOk queueBind(String queue, String exchange,
        // String routingKey) throws IOException;
　　　　　// 注意绑定支持一个queue绑定多个routingKey，或者一个routingKey绑定到多个queue
        queueBind("vinoQueue", "vino", "hello");
        // 发布消息
        String message = "Hello Wrold " + Math.random();
　　　　　// 第一个参数是exchange的名称，若用""表示发送到默认的exchange，rabbitmq的默认的exchange的类型是direct。
        channel.basicPublish("vino", "hello", null, message.getBytes());

剩下就是最后一件事（生产者和消费者都需要调用的）

channel使用完毕后要关闭

channel.close();
conn.close();

参考：兔子和兔子窝

优质内容筛选与推荐>>
1、SQL升级脚本实现按版本差异化升级(优化)
2、Aspx折线图、柱状图示例（Web Chart）
3、c#中邮件收发处理（POP3,IMAP,SMTP）
4、Android DrawerLayout 高仿QQ5.2双向侧滑菜单
5、中国剩余定理

0 │ 收藏 │ 举报

朋友将在看一看看到

分享想法到看一看