Java学习——rabbitmq(routing)

Routing

/* Routing Model
                   /(friends) [...](Q1) -> friend consumers
P -> X(type direct)
                   \(enemies) [...](Q2) -> enemy consumers
*/

前面的Pub/Sub模型实现将消息发布给所有监听这个队列的消费者，如果要指定某些消息发布给特定的几个消费者，需要用到Routing模型。

 

记得Pub/Sub中我们绑定exchange和queue用到下面的代码

 

channel.queueBind(queueName, "chatroom", "");

这里有第三个参数，它的含义依赖于exchange的类型，如果是fanout类型，则这个参数被忽略。fanout类型不容易定制，我们下面用direct类型来实现，此时queue会根据第三个参数来向exchange获取消息。

 

channel.exchangeDeclare("chatroom","direct");
channel.basicPublish("chatroom","friends",null,mesage.getBytes());

//和Pub/Sub一样，我们这里需要一个临时队列
String queueName = channel.queueDeclare().getQueue();
channel.queueBind(queueName, "chatroom", "friends");

Publisher: Send.java

 

package test;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class Send {

	public static void main(String[] args) throws IOException {
		// 创建一个连接连接服务器
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		
		channel.exchangeDeclare("chatroom","direct");
		
		for(int i = 1; i < 5; i ++){
			String message = "message " + i;
			if(i%2 == 0){
				channel.basicPublish("chatroom", "friends", null, message.getBytes());
				System.out.println(" [x] Sent '" + "friends" + "':'" + message +"'");
			}
			else{
				channel.basicPublish("chatroom", "enemies", null, message.getBytes());
				System.out.println(" [x] Sent '" + "enemies" + "':'" + message +"'");
			}
		}
		channel.close();
		connection.close();
	}
}

Subscriber: Recv.java

 

package test;

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ShutdownSignalException;

public class Recv {

	public static void main(String[] args) throws IOException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {
		// 创建一个连接接收数据
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost("localhost");
		//factory.setPort(1987);
		Connection connection = factory.newConnection();
		Channel channel = connection.createChannel();
		
		channel.exchangeDeclare("chatroom","direct");
		String queueName = channel.queueDeclare().getQueue();
		channel.queueBind(queueName, "chatroom", "friends");
		channel.queueBind(queueName, "chatroom", "enemies");
		// 等待消息
		System.out.println("waiting for messages from all");
		QueueingConsumer consumer = new QueueingConsumer(channel);
		channel.basicConsume(queueName, true, consumer);
		while(true){
			QueueingConsumer.Delivery delivery = consumer.nextDelivery();
			String message = new String(delivery.getBody());
			System.out.println(" [x] Received '" + message + "'");
		}
	}
}

执行效果如下：（开启3个Recv）

 

//Recv 1.java
channel.queueBind(queueName, "chatroom", "friends");
System.out.println("waiting for messages from friends");
/* output
waiting for messages from friends
 [x] Received 'message 2'
 [x] Received 'message 4'
*/
//Recv 2.java
channel.queueBind(queueName, "chatroom", "enemies");
System.out.println("waiting for messages from enemies");
/* output
waiting for messages from enemies
 [x] Received 'message 1'
 [x] Received 'message 3'
*/
//Recv 3.java
channel.queueBind(queueName, "chatroom", "friends");
channel.queueBind(queueName, "chatroom", "enemies");
System.out.println("waiting for messages from all");
/* output
waiting for messages from all
 [x] Received 'message 1'
 [x] Received 'message 2'
 [x] Received 'message 3'
 [x] Received 'message 4'
*/

由于前面的pub/sub用到了chatroom这个exchange，所以需要做如下处理

 

// cannot redeclare exchange 'chatroom' in vhost '/' with different type
root > rabbitmqctl.bat list_queues
Listing queues ...
Error: unable to connect to node rabbit@ciaos-desktop: nodedown

windows下解决上面的错误的方法如下

 

cp -r C:\Windows\.erlang.cookie C:\Users\p00206869\.erlang.cookie

root > rabbitmqctl.bat list_exchanges
Listing exchanges ...
	direct
amq.direct	direct
amq.fanout	fanout
amq.headers	headers
amq.match	headers
amq.rabbitmq.log	topic
amq.rabbitmq.trace	topic
amq.topic	topic
chatroom	fanout
...done.

可以看到chatroom类型为fanout，重置rabbitmq服务（还不知道什么命令可以删除exchange）

 

rabbitmqctl.bat stop_app
rabbitmqctl.bat reset
rabbitmqctl.bat start_app

重启rabbitmq服务的方法如下

 

rabbitmq-service.bat stop
rabbitmq-service.bat start