package com.shi.topic;import java.io.IOException;
import java.util.concurrent.TimeoutException;import org.junit.Test;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.QueueingConsumer.Delivery;
import com.shi.util.RabbitMqUtils;/*** 通配符模式  - topic* @author SHF* @version 创建时间：2018年7月4日  下午5:53:29*/
public class TopicTest {//交换机名称private final static String EXCHANGE_NAME = "exchange_topic";//路由 keyprivate final static String KEY_1 ="a.1";private final static String KEY_2 ="key.*";private final static String KEY_3 ="#.#";//队列名称private final static String QUEUE_1 ="queue_topic_1";private final static String QUEUE_2 ="queue_topic_2";/*** 生产者 - 路由模式* KEY_1 ="a";* @author SHF* @version 创建时间：2018年7月4日  下午4:20:39* @throws TimeoutException * @throws IOException */@Testpublic void send()throws IOException, TimeoutException {//1 获取链接及mq 通道Connection connection = RabbitMqUtils.getConnection();Channel channel = connection.createChannel();//2 声明exchangechannel.exchangeDeclare(EXCHANGE_NAME, "topic");//3 消息内容String message = " 施爷 通配符模式 topic 向你发送了一条消息....";channel.basicPublish(EXCHANGE_NAME, KEY_1, null, message.getBytes());System.out.println(" [x] sent:"+message);//4关闭通道及连接channel.close();connection.close();}/*** 消费者1 - 路由模式* KEY_2 ="b";* @author SHF* @version 创建时间：2018年7月4日  下午4:33:55* @throws TimeoutException * @throws IOException * @throws InterruptedException * @throws ConsumerCancelledException * @throws ShutdownSignalException */@Testpublic void reic1() throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {//1 获取连接 及 通道Connection connection = RabbitMqUtils.getConnection();Channel channel = connection.createChannel();//2 声明队列channel.queueDeclare(QUEUE_1, false, false, false, null);//3 绑定交换机,指定路由channel.queueBind(QUEUE_1, EXCHANGE_NAME, KEY_2);//4 同一个服务器只会发送一条消息给消费者channel.basicQos(1);//5 定义队列的消费者QueueingConsumer consumer = new QueueingConsumer(channel);//6 监听队列，手动返回完成channel.basicConsume(QUEUE_1, false,consumer);//7 获取消息while(true) {Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());System.out.println( "[x] reiv1 :" + message);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}/*** 消费者2 - 路由模式* KEY_3 ="a"* @author SHF* @version 创建时间：2018年7月4日  下午4:33:55* @throws TimeoutException * @throws IOException * @throws InterruptedException * @throws ConsumerCancelledException * @throws ShutdownSignalException */@Testpublic void reic2() throws IOException, TimeoutException, ShutdownSignalException, ConsumerCancelledException, InterruptedException {//1 获取连接 及 通道Connection connection = RabbitMqUtils.getConnection();Channel channel = connection.createChannel();//2 声明队列channel.queueDeclare(QUEUE_2, false, false, false, null);//3 绑定交换机,指定路由channel.queueBind(QUEUE_2, EXCHANGE_NAME, KEY_3);//4 同一个服务器只会发送一条消息给消费者channel.basicQos(1);//5 定义队列的消费者QueueingConsumer consumer = new QueueingConsumer(channel);//6 监听队列，手动返回完成channel.basicConsume(QUEUE_2, false,consumer);//7 获取消息while(true) {Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());System.out.println( "[x] reiv2 :" + message);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}
}