5.rabbitmq--通配符模式Topics
rabbitmq--通配符模式Topics
topic模式也称为主题模式，其实他相对于routing模式最大的好处就是他多了一种匹配模式的路由，怎么理解匹配呢，其实就至关于咱们以前正则的.*这种，不过他的匹配机制可能不是这种（其实除了匹配规则外，他的做用就和routing模式同样 ），而他的工做流程图以下：java
OK！ 先说一下他的匹配规则：web
绑定键binding key也必须是这种形式。以特定路由键发送的消息将会发送到全部绑定键与之匹配的队列中。但绑定键有两种特殊的状况：
①*（星号）仅表明一个单词
②#（井号）表明任意个单词spring
示例：apache
以上图为例：app
*.orange.* : 匹配以 任意一个单词字符开头中间包含 .orange. 以任意一个单词字符结尾 的字符串。好比 a.orange.b, sdfsd.orange.fdsfsdf 等（注意是一个单词）。ide
lay.# ：只要一lay.开头的都匹配，他能够匹配lay.a, lay.a.b, lay.b.c等。fetch
这样是否是很方便，好比咱们想将log的发给q1队列，其余的发给q2，那么咱们只须要定义log.#、或者log.*,那么你发送给q1队列的数据就是log日志的消息。 spa
1 package com.maozw.mq.topic; 2 3 import com.maozw.mq.config.RabbitConfig; 4 import com.rabbitmq.client.Channel; 5 import org.slf4j.Logger; 6 import org.slf4j.LoggerFactory; 7 import org.springframework.amqp.rabbit.connection.Connection; 8 import org.springframework.amqp.rabbit.connection.ConnectionFactory; 9 import org.springframework.beans.factory.annotation.Autowired; 10 import org.springframework.web.bind.annotation.RequestMapping; 11 import org.springframework.web.bind.annotation.RestController; 12 13 import java.io.IOException; 14 import java.util.concurrent.TimeoutException; 15 16 import static org.apache.log4j.varia.ExternallyRolledFileAppender.OK; 17 18 /** 19 * work 模式 20 * 两种分发： 轮询分发 + 公平分发 21 * 轮询分发：消费端：自动确认消息；boolean autoAck = true; 22 * 公平分发: 消费端：手动确认消息 boolean autoAck = false; channel.basicAck(envelope.getDeliveryTag(),false); 23 * 24 * @author MAOZW 25 * @Description: 商品 26 * @date 2018/11/26 15:06 27 */ 28 @RestController 29 @RequestMapping("/topic") 30 public class TopicProducer { 31 private static final Logger LOGGER = LoggerFactory.getLogger(TopicProducer.class); 32 @Autowired 33 RabbitConfig rabbitConfig; 34 35 public static final String EXCHANGE_TOPIC = "EXCHANGE_TOPIC"; 36 private static final String[] routingKeys = {"commodity.add", "commodity.update", "commodity.select", "commodity.delete"}; 37 38 @RequestMapping("/send") 39 public String send() throws IOException, TimeoutException { 40 Connection connection = null; 41 Channel channel = null; 42 try { 43 ConnectionFactory connectionFactory = rabbitConfig.connectionFactory(); 44 connection = connectionFactory.createConnection(); 45 channel = connection.createChannel(false); 46 47 /** 48 * 申明交换机 以及type 49 */ 50 channel.exchangeDeclare(EXCHANGE_TOPIC, "topic"); 51 /** 52 * 发送消息 53 * 每一个消费者 发送确认消息以前，消息队列不会发送下一个消息给消费者，一次只处理一个消息 54 * 自动模式无需设置下面设置 55 */ 56 int prefetchCount = 1; 57 channel.basicQos(prefetchCount); 58 59 String Hello = ">>>> Hello EXCHANGE_TOPIC <<<<"; 60 String message = Hello; 61 for (String routingKey : routingKeys) { 62 channel.basicPublish(EXCHANGE_TOPIC, routingKey, null, message.getBytes()); 63 } 64 LOGGER.info("生产消息： " + message); 65 return "OK"; 66 } catch (Exception e) { 67 68 } finally { 69 connection.close(); 70 channel.close(); 71 return OK; 72 } 73 } 74 }
1 package com.maozw.mq.topic; 2 3 import com.maozw.mq.config.RabbitConfig; 4 import com.rabbitmq.client.AMQP; 5 import com.rabbitmq.client.Channel; 6 import com.rabbitmq.client.DefaultConsumer; 7 import com.rabbitmq.client.Envelope; 8 import org.slf4j.Logger; 9 import org.slf4j.LoggerFactory; 10 import org.springframework.amqp.rabbit.connection.Connection; 11 import org.springframework.amqp.rabbit.connection.ConnectionFactory; 12 13 import java.io.IOException; 14 15 /** 16 * @author MAOZW 17 * @Description: ${todo} 18 * @date 2018/11/26 15:06 19 */ 20 21 public class TopicConsumer { 22 private static final Logger LOGGER = LoggerFactory.getLogger(TopicConsumer.class); 23 public static final String EXCHANGE_TOPIC = "EXCHANGE_TOPIC"; 24 private static final String[] routingKeys = {"commodity.add", "commodity.update", "commodity.delete"}; 25 26 public static void main(String[] args) throws IOException { 27 ConnectionFactory connectionFactory = RabbitConfig.getConnectionFactory(); 28 Connection connection = connectionFactory.createConnection(); 29 Channel channel = connection.createChannel(false); 30 /** 31 * 建立队列申明 32 */ 33 boolean durable = true; 34 channel.queueDeclare(RabbitConfig.QUEUE_TOPIC, durable, false, false, null); 35 /** 36 * 绑定队列到交换机 37 */ 38 channel.queueBind(RabbitConfig.QUEUE_TOPIC, EXCHANGE_TOPIC, "commodity.add"); 39 40 41 /** 42 * 改变分发规则 43 */ 44 channel.basicQos(1); 45 DefaultConsumer consumer = new DefaultConsumer(channel) { 46 @Override 47 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 48 super.handleDelivery(consumerTag, envelope, properties, body); 49 System.out.println("[1] 接口数据 ： " + new String(body, "utf-8")); 50 try { 51 Thread.sleep(300); 52 } catch (InterruptedException e) { 53 e.printStackTrace(); 54 } finally { 55 System.out.println("[1] done!"); 56 //消息应答：手动回执，手动确认消息 57 channel.basicAck(envelope.getDeliveryTag(), false); 58 } 59 } 60 }; 61 //监听队列 62 /** 63 * autoAck 消息应答 64 * 默认轮询分发打开：true ：这种模式一旦rabbitmq将消息发送给消费者，就会从内存中删除该消息，不关心客户端是否消费正常。 65 * 使用公平分发须要关闭autoAck:false 须要手动发送回执 66 */ 67 boolean autoAck = false; 68 channel.basicConsume(RabbitConfig.QUEUE_TOPIC, autoAck, consumer); 69 } 70 71 }
1 package com.maozw.mq.topic; 2 3 import com.maozw.mq.config.RabbitConfig; 4 import com.rabbitmq.client.AMQP; 5 import com.rabbitmq.client.Channel; 6 import com.rabbitmq.client.DefaultConsumer; 7 import com.rabbitmq.client.Envelope; 8 import org.slf4j.Logger; 9 import org.slf4j.LoggerFactory; 10 import org.springframework.amqp.rabbit.connection.Connection; 11 import org.springframework.amqp.rabbit.connection.ConnectionFactory; 12 13 import java.io.IOException; 14 15 /** 16 * @author MAOZW 17 * @Description: ${todo} 18 * @date 2018/11/26 15:06 19 */ 20 21 public class TopicConsumer2 { 22 private static final Logger LOGGER = LoggerFactory.getLogger(TopicConsumer2.class); 23 public static final String EXCHANGE_TOPIC = "EXCHANGE_TOPIC"; 24 private static final String[] routingKeys = {"commodity.add", "commodity.update", "commodity.delete", "commodity.select"}; 25 26 public static void main(String[] args) throws IOException { 27 ConnectionFactory connectionFactory = RabbitConfig.getConnectionFactory(); 28 Connection connection = connectionFactory.createConnection(); 29 Channel channel = connection.createChannel(false); 30 /** 31 * 建立队列申明 32 */ 33 boolean durable = true; 34 channel.queueDeclare(RabbitConfig.QUEUE_TOPIC2, durable, false, false, null); 35 36 /** 37 * 绑定队列到交换机 38 */ 39 channel.queueBind(RabbitConfig.QUEUE_TOPIC2, EXCHANGE_TOPIC, "commodity.*"); 40 41 42 /** 43 * 改变分发规则 44 */ 45 channel.basicQos(1); 46 DefaultConsumer consumer = new DefaultConsumer(channel) { 47 @Override 48 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 49 super.handleDelivery(consumerTag, envelope, properties, body); 50 System.out.println("[2] 接口数据 ： " + new String(body, "utf-8")); 51 try { 52 Thread.sleep(200); 53 } catch (InterruptedException e) { 54 e.printStackTrace(); 55 } finally { 56 System.out.println("[2] done! "); 57 //消息应答：手动回执，手动确认消息 58 channel.basicAck(envelope.getDeliveryTag(), false); 59 } 60 } 61 }; 62 //监听队列 63 /** 64 * autoAck 消息应答 65 * 默认轮询分发打开：true ：这种模式一旦rabbitmq将消息发送给消费者，就会从内存中删除该消息，不关心客户端是否消费正常。 66 * 使用公平分发须要关闭autoAck:false 须要手动发送回执 67 */ 68 boolean autoAck = false; 69 channel.basicConsume(RabbitConfig.QUEUE_TOPIC2, autoAck, consumer); 70 } 71 }