Java学习——rabbitmq(topic)

Topic

/* Routing Model
                   /(*.friends) [...](Q1) -> consumers one
P -> X(type topic)
                   \(*.enemies|female.#) [...](Q2) -> consumers two
*/

前面的Routing虽然可以把消息分组路由给不同的消费群体（consumers），但是消费者只能监听到这个生产者对应key的消息，如何让它同时再监听别的生产者的消息，我们这里试用一下Topic模型。

 

Topic Exchange 和 Direct Exchange很类似，不过它的key支持简单的正则匹配

 

* (star) can substitute for exactly one word

# (hash) can substiture for zero or more words

 

C1对所有(男|女)朋友的话都感兴趣，C2对(男|女)敌人的话和女(朋友|敌人)的话感兴趣，这样女性朋友的话会被同时分发给C1和C2，这是在Routing模型下不能实现的。

topic类型比较灵活，当一个队列bind一个"#"的key时，它接收所有消息，此时exchange类型就像fanout一样，同时，如果我们没有使用"*"或者"#"，则exchange类型类似direct类型。

 

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","topic");
		
		String message = "hi all friends";
		channel.basicPublish("chatroom", "*.friends", null, message.getBytes());
		System.out.println(" [x] Sent '" + "*.friends" + "':'" + message +"'");
		
		message = "bye all enemies";
		channel.basicPublish("chatroom", "*.enemies", null, message.getBytes());
		System.out.println(" [x] Sent '" + "*.enemies" + "':'" + message +"'");
		
		message = "delete all females friends or enemies";
		channel.basicPublish("chatroom", "female.#", null, message.getBytes());
		System.out.println(" [x] Sent '" + "female.#" + "':'" + message +"'");
		
		channel.close();
		connection.close();
	}
}

Consumer 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","topic");
		String queueName = channel.queueDeclare().getQueue();
		channel.queueBind(queueName, "chatroom", "*.enemies");
		channel.queueBind(queueName, "chatroom", "female.*");
		// 等待消息
		System.out.println("waiting for messages from enemies and females");
		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 + "'");
		}
	}
}

 运行效果如下：

 

//Recv 1.java
channel.queueBind(queueName, "chatroom", "*.friends");
System.out.println("waiting for messages from friends");
/* output
waiting for messages from all
 [x] Received 'hi all friends'
*/
//Recv 2.java
channel.queueBind(queueName, "chatroom", "*.enemies");
channel.queueBind(queueName, "chatroom", "female.*");
System.out.println("waiting for messages from enemies and females");
/* output
waiting for messages from enemies and females
 [x] Received 'bye all enemies'
 [x] Received 'delete all females friends or enemies'
*/

使用topic和正则可以减少很多队列key，比起routing要灵活很多