rabbitmq 发布／订阅

工作队列，每个任务只分发给一个工作者（worker）。http://blog.csdn.net/convict_eva/article/details/52292845

“发布／订阅”：分发一个消息给多个消费者（consumers）。

准备

    构建一个简单的日志系统。发布者生产日志，消费者接收日志。
交换机

    让我们简单的概括一下之前的教程：

     1、发布者（producer）是发布消息的应用程序。

     2、队列（queue）用于消息存储的缓冲。

    3、消费者（consumer）是接收消息的应用程序。

    RabbitMQ消息模型的核心理念是：发布者（producer）不会直接发送任何消息给队列。事实上，发布者（producer）甚至不知道消息是否已经被投递到队列。

    发布者（producer）只需要把消息发送给一个交换机（exchange）。交换机非常简单，它一边从发布者方接收消息，一边把消息推送到队列。

    交换机必须知道如何处理它接       收到的消息，是应该推送到指定的队列还是多个队列，或者是直接忽略消息。这些规则是通过交换机类型（exchange type）来定义的。

    交换机类型：直连交换机（direct）, 主题交换机（topic）, 头交换机（headers）和 扇型交换机（fanout）。    这时主要说明扇型交换机，它把消息发送给它所知道的所有队列。

    上两个文章（http://blog.csdn.net/convict_eva/article/details/52292845，http://blog.csdn.net/convict_eva/article/details/52291774）中没有使用到交换机（没有指定交换机名称），代码：channel.basicPublish("", "task_queue", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());第一个参数就是交换机名称，空字符串代表默认或者匿名交换机，消息将会根据指定的routing_key分发到指定的队列。

    

    查看交换机列表：./rabbitmqctl list_exchanges

    
临时队列

    如果打算在发布者（producers）和消费者（consumers）之间共享同队列的话，给队列命名是十分重要的。

    日志系统关心的是最新的日志，并且要接收所有的日志，那么指定队列名称就不合适。要实现这个目的要做两件事：

    1、当连接上rabbitmq 时需要一个全新的、空的队列，可以手动随机创建一个队列名或者让服务器为我们选择一个随机队列（推荐）。只要在声明队列时不指定队列名即可。

    2、当与消费者（consumer）断开连接的时候，这个队列应当被立即删除。exclusive标识符即可达到此目的。

    代码如下：

    //服务器指定随机队列,不指定队列名称，不待久化，当前连接独有的（EXCLUSIVE），自动删除

    String queueName = channel.queueDeclare("",RabbitmqConfigure.DURABLE,RabbitmqConfigure.EXCLUSIVE,RabbitmqConfigure.AUTOD_ELETE,null).getQueue();

    接收消息时，使用此队列名称：channel.basicConsume(queueName,  RabbitmqConfigure.AUTOACK, consumer);

    
绑定

    我们已经创建了一个扇型交换机（fanout）和一个队列。现在我们需要告诉交换机如何发送消息给我们的队列。交换器和队列之间的联系我们称之为绑定（binding）。可以理解为：这个队列（queue）对这个交换机（exchange）的消息感兴趣。

    code:

    消费者队列绑定到exchange

    //把队列和交换机绑定

    channel.queueBind(queueName, RabbitmqConfigure.EXCHANGE_NAME, "");

    生产者声明exchange

    //声明交换机

    channel.exchangeDeclare(RabbitmqConfigure.EXCHANGE_NAME, RabbitmqConfigure.EXCHANGE_TYPE);

    

    查看绑定列表：

        命令行#./rabbitmqctl list_bindings，也可心通过管理页面查看。如图2.png

    
代码说明

文件如下：



        RabbitmqConfigure    配置

package com.convict.rabbitmq.publishsubscribe;
public class RabbitmqConfigure {
//队列名称
public final static String QUEUE_NAME = "publishsubscribe_queue";
//是否是持久化的queue
public final static boolean DURABLE=false;
//受当前连接限制的queue，true：当与消费者（consumer）断开连接的时候，这个队列应当被立即删除。
public final static boolean EXCLUSIVE=true;
//是否自动删除，如果长时间没有用到自动删除
public final static boolean AUTOD_ELETE=true;
//自动确认消息,   false 不自动确认，要手动确认消息
public final static boolean AUTOACK = true;
public final static String HOST = "192.168.174.128";
//用户要提前创建
public final static String PASS_WORD="convict_eva";
public final static String USER_NAME="convict_eva";
// virtual host 要提前创建  /convict_eva
public final static String VIRTUAL_HOST="/convict_eva";
//交换机名称
public final static String EXCHANGE_NAME="logs";
public final static String EXCHANGE_TYPE="fanout";
}

Recv    消费者

package com.convict.rabbitmq.publishsubscribe;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.Map;

public class Recv {
public static void main(String[] args)  throws Exception{

//设置和send 相同，
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(RabbitmqConfigure.HOST);
factory.setPassword(RabbitmqConfigure.PASS_WORD);
factory.setUsername(RabbitmqConfigure.USER_NAME);
factory.setVirtualHost(RabbitmqConfigure.VIRTUAL_HOST);
//打开connection 和 channel
Connection connection = factory.newConnection();
final Channel channel = connection.createChannel();
//声明交换机
channel.exchangeDeclare(RabbitmqConfigure.EXCHANGE_NAME, RabbitmqConfigure.EXCHANGE_TYPE);
//服务器指定随机队列,不指定队列名称，不待久化，当前连接独有的（EXCLUSIVE），自动删除
String queueName = channel.queueDeclare("",RabbitmqConfigure.DURABLE,RabbitmqConfigure.EXCLUSIVE,
RabbitmqConfigure.AUTOD_ELETE,null).getQueue();

System.out.println("random queue name is :" + queueName);

//把队列和交换机绑定
channel.queueBind(queueName, RabbitmqConfigure.EXCHANGE_NAME, "");

//公平调度：告诉RabbitMQ，再同一时刻，不要发送超过1条消息给一个工作者（worker），直到它已经处理了上一条消息并且作出了响应。
//这样，RabbitMQ就会把消息分发给下一个空闲的工作者（worker）。
channel.basicQos(1);
System.out.println("Waiting for messages.");
/**
* consumer 接收消息回调方法，DefaultConsumer提供一个方法可以缓存发送的消息，直到消息被消费。
*/
final Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
throws IOException {
String message = new String(body, "UTF-8");
System.out.println("Received '" + message + "'");
try {
doWork(message);
}catch(Exception e){
System.err.println("执行任务异常！！！");
}
}
};
//自动确认，队列名称为服务器指定的队列名称
channel.basicConsume(queueName,  RabbitmqConfigure.AUTOACK, consumer);
}
private static void doWork(String task) throws InterruptedException {
for (char ch: task.toCharArray()) {
if (ch == '.') Thread.sleep(1000);
}
}
}

        RecvB    Recv复制一份，模拟两个消费者

        Send    生产者

package com.convict.rabbitmq.publishsubscribe;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
public class Send {
public static void main(String[] argv) throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(RabbitmqConfigure.HOST);
factory.setPassword(RabbitmqConfigure.PASS_WORD);
factory.setUsername(RabbitmqConfigure.USER_NAME);
//VirtualHost 要在控制台提前创建
factory.setVirtualHost(RabbitmqConfigure.VIRTUAL_HOST);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

//声明交换机
channel.exchangeDeclare(RabbitmqConfigure.EXCHANGE_NAME, RabbitmqConfigure.EXCHANGE_TYPE);

for(int i=0;i<10;i++){
String message = "hello...."+i;
//往交换机中发送消息，而不是往队列中发送消息
channel.basicPublish(RabbitmqConfigure.EXCHANGE_NAME, "", null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");
}
channel.close();
connection.close();
}
}

测试：

    先启动消费者

    1、启动Recv，控制台打印的队列名称：random queue name is :amq.gen-eccKxDOzkyJakl9O9Lb3iw

    2、启动Send，发送消息

    消费两个消息后强制停止Recv， 

    再次启动Recv 就没有消息可消费了。重启Recv后，是新的queue和exchange 绑定，之前的queue被删除，发送到queue的消息也就丢失了。

    先启动生产者发送消息，再启动消费者时没有消息可消费，因为没有queue和exchange绑定，exchange 把消息忽略了，queue和exchange 绑定后，再往exchange发送消息时exchange 会把消息发送到绑定的队列中。

rabbitmq 官网：http://www.rabbitmq.com/tutorials/tutorial-three-java.html

rabbitmq 中文：http://rabbitmq.mr-ping.com/tutorials_with_python/[3]Publish_Subscribe.html

代码：http://download.csdn.net/detail/convict_eva/9611560