系列： http://zxb1985.iteye.com/category/267524
在前面的已经提到了一对一的情况;现在一个生产者与多个消费者的情况（Work Queues）。
Work Queues的示意图如下：
对于上图的模型中对于c端的worker来说。RabbitMQ服务器可能一直发送多个消息给一个worker，而另一个可能几乎不做任何事情。这样就会导致一个worker很忙，而另一个却很空闲。这种情况可能都不想出现。如何解决这个问题呢。当然最理想的情况是均匀分配消息给每个worker。我们可能通过channel.basicQos(1)方法（prefetchCount=1）来设置同一时间每次发给一个消息给一个worker。示意图如下：
P端的程序如下：
packagecom.abin.rabbitmq;
importcom.rabbitmq.client.Channel;
importcom.rabbitmq.client.Connection;
importcom.rabbitmq.client.ConnectionFactory;
importcom.rabbitmq.client.MessageProperties;
publicclassNewTask?{
privatestaticfinalString?TASK_QUEUE_NAME?="task_queue";
publicstaticvoidmain(String[]?argv)throwsException?{
ConnectionFactory?factory?=newConnectionFactory();
factory.setHost("localhost");
Connection?connection?=?factory.newConnection();
Channel?channel?=?connection.createChannel();
//声明此队列并且持久化
channel.queueDeclare(TASK_QUEUE_NAME,true,false,false,null);
String?message?=?getMessage(argv);
channel.basicPublish("",?TASK_QUEUE_NAME,
MessageProperties.PERSISTENT_TEXT_PLAIN,?message.getBytes());//持久化消息
System.out.println("?[x]?Sent?'"+?message?+"'");
channel.close();
connection.close();
}
privatestaticString?getMessage(String[]?strings)?{
if(strings.length?<1)
return"Hello?World!";
returnjoinStrings(strings,"?");
}
privatestaticString?joinStrings(String[]?strings,?String?delimiter)?{
intlength?=?strings.length;
if(length?==0)
return"";
StringBuilder?words?=newStringBuilder(strings[0]);
for(inti?=1;?i?<?length;?i++)?{
words.append(delimiter).append(strings[i]);
}
returnwords.toString();
}
}
多次运行此程序并传入的参数分别为“First message”，“Secondmessage”，“Thirdmessage”，“Fourth message”，“Fifth message”
C端的程序如下：
Java代
packagecom.abin.rabbitmq;
importcom.rabbitmq.client.Channel;
importcom.rabbitmq.client.Connection;
importcom.rabbitmq.client.ConnectionFactory;
importcom.rabbitmq.client.QueueingConsumer;
publicclassWorker?{
privatestaticfinalString?TASK_QUEUE_NAME?="task_queue";
publicstaticvoidmain(String[]?argv)throwsException?{
ConnectionFactory?factory?=newConnectionFactory();
factory.setHost("localhost");
Connection?connection?=?factory.newConnection();
Channel?channel?=?connection.createChannel();
//声明此队列并且持久化
channel.queueDeclare(TASK_QUEUE_NAME,true,false,false,null);
System.out.println("?[*]?Waiting?for?messages.?To?exit?press?CTRL+C");
channel.basicQos(1);//告诉RabbitMQ同一时间给一个消息给消费者
/*?We're?about?to?tell?the?server?to?deliver?us?the?messages?from?the?queue.
*?Since?it?will?push?us?messages?asynchronously,
*?we?provide?a?callback?in?the?form?of?an?object?that?will?buffer?the?messages
*?until?we're?ready?to?use?them.?That?is?what?QueueingConsumer?does.*/
QueueingConsumer?consumer?=newQueueingConsumer(channel);
/*
把名字为TASK_QUEUE_NAME的Channel的值回调给QueueingConsumer,即使一个worker在处理消息的过程中停止了，这个消息也不会失效
*/
channel.basicConsume(TASK_QUEUE_NAME,false,?consumer);
while(true)?{
QueueingConsumer.Delivery?delivery?=?consumer.nextDelivery();//得到消息传输信息
String?message?=newString(delivery.getBody());
System.out.println("?[x]?Received?'"+?message?+"'");
doWork(message);
System.out.println("?[x]?Done");
channel.basicAck(delivery.getEnvelope().getDeliveryTag(),false);//下一个消息
}
}
privatestaticvoiddoWork(String?task)throwsInterruptedException?{
for(charch?:?task.toCharArray())?{
if(ch?=='.')
Thread.sleep(1000);//这里是假装我们很忙
}
}
}
开启两个worker分别运行。运行结果如：
c1的结果：
Java代码
[*]?Waitingformessages.?To?exit?press?CTRL+C
[x]?Received'First?message'
[x]?Received'Third?message'
[x]?Received'Fifth?message'
c2的结果
[*]?Waitingformessages.?To?exit?press?CTRL+C
[x]?Received'Second?message'
[x]?Received'Fourth?message'
在前面的Work Queue中的消息是均匀分配消息给消费者；如果我想把消息分发给所有的消费者呢？那应当怎么操作呢？这就是要下面提到的Publish/Subscribe（分布/订阅）。让我们开始Publish/Subscribe之旅吧!
Publish/Subscribe的工作示意图如下：
在上图中的X表示Exchange(交换区);Exchange的类型有：direct,topic,headers和fanout
Publish/Subscribe的Exchang的类型为fanout;声明Publish/Subscribe的Exchang代码如下：
channel.exchangeDeclare("logs","fanout");
对于Work Queue中提到的发布消息的代码如下：
Java代码
channel.basicPublish("",?queueName,null,?message.getBytes());
但对于Publish/Subscribe中发布消息中的Queue的使用的是默认的；代码如下：
Java代码
channel.basicPublish("logs","",null,?message.getBytes());
Exchange和各Queue之间是如何通信的呢？主要是通过把Exchange和各Queue绑定（binding）；示意代码如下:
Java代码
channel.queueBind(queueName,?exchangeName,"");
Publish/Subscribe加入绑定的工作示意图如下：
那我们就开始程序代码吧：P端的代码如下：
Java代码
packagecom.abin.rabbitmq;
importcom.rabbitmq.client.Channel;
importcom.rabbitmq.client.Connection;
importcom.rabbitmq.client.ConnectionFactory;
publicclassEmitLog?{
privatestaticfinalString?EXCHANGE_NAME?="logs";
publicstaticvoidmain(String[]?argv)throwsException?{
ConnectionFactory?factory?=newConnectionFactory();
factory.setHost("localhost");
Connection?connection?=?factory.newConnection();
Channel?channel?=?connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");//声明Exchange
for(inti?=0;?i?<=2;?i++)?{
String?message?="hello?word!"+?i;
channel.basicPublish(EXCHANGE_NAME,"",null,?message.getBytes());
System.out.println("?[x]?Sent?'"+?message?+"'");
}
channel.close();
connection.close();
}
}
运行结果如下：
[x]?Sent'hello?word!0'
[x]?Sent'hello?word!1'
[x]?Sent'hello?word!2'
C端的代码如下：
Java代码
packagecom.abin.rabbitmq;
importcom.rabbitmq.client.Channel;
importcom.rabbitmq.client.Connection;
importcom.rabbitmq.client.ConnectionFactory;
importcom.rabbitmq.client.QueueingConsumer;
publicclassReceiveLogsOne?{
privatestaticfinalString?EXCHANGE_NAME?="logs";
publicstaticvoidmain(String[]?argv)throwsException?{
ConnectionFactory?factory?=newConnectionFactory();
factory.setHost("localhost");
Connection?connection?=?factory.newConnection();
Channel?channel?=?connection.createChannel();
channel.exchangeDeclare(EXCHANGE_NAME,"fanout");
String?queueName?="log-fb1";
channel.queueDeclare(queueName,false,false,false,null);
channel.queueBind(queueName,?EXCHANGE_NAME,"");//把Queue、Exchange绑定
QueueingConsumer?consumer?=newQueueingConsumer(channel);
channel.basicConsume(queueName,true,?consumer);
while(true)?{
QueueingConsumer.Delivery?delivery?=?consumer.nextDelivery();
String?message?=newString(delivery.getBody());
System.out.println("?[x]?Received?'"+?message?+"'");
}
}
}
对于C端的代码我写了二个差不多的程序，只需要修改一下queueName。这样就形成了二个Queue;运行结果相同；
运行结果可能如下：
Java代
[x]?Received'hello?word!0'
[x]?Received'hello?word!1'
[x]?Received'hello?word!2'
