RabbitMq七种工作模式，结合简单的java实例使用，别再说你不会( 五 ) _RabbitMq

5.1.2、Receiver1.javaimport com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.QueueingConsumer;import top.san.RabbitMq.util.RabbitmqConnectionUtil;import java.io.IOException;public class Receiver1 {private final static String QUEUE_NAME = "queue_topic";private final static String EXCHANGE_NAME = "exchange_topic";private final static String EXCHANGE_TYPE = "topic";public static void main(String[] args) throws IOException, InterruptedException {Connection connection = RabbitmqConnectionUtil.getConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false,false, null);//第二参数就是去匹配我兴趣的topicchannel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "lbw.nb.*");channel.basicQos(1);QueueingConsumer consumer = new QueueingConsumer(channel);channel.basicConsume(QUEUE_NAME, false, consumer);while(true){QueueingConsumer.Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());System.out.println("[x] Received1 '"+message + "'");Thread.sleep(10);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}}5.1.3、Receiver2.javaimport com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.QueueingConsumer;import top.san.RabbitMq.util.RabbitmqConnectionUtil;import java.io.IOException;public class Receiver2 {private final static String QUEUE_NAME = "queue_topic2";private final static String EXCHANGE_NAME = "exchange_topic";private final static String EXCHANGE_TYPE = "topic";public static void main(String[] args) throws IOException, InterruptedException {Connection connection = RabbitmqConnectionUtil.getConnection();Channel channel = connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false,false, null);//第二参数就是去匹配我兴趣的topicchannel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "lbw.#");channel.basicQos(1);QueueingConsumer consumer = new QueueingConsumer(channel);channel.basicConsume(QUEUE_NAME, false, consumer);while(true){QueueingConsumer.Delivery delivery = consumer.nextDelivery();String message = new String(delivery.getBody());System.out.println("[x] Received2 '"+message + "'");Thread.sleep(10);channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);}}}5.1.4、结果-分析话题的特色就是队列可以获取自己感兴趣的话题消息，可以通过通配符*或#来表示匹配所有的感兴趣的字符串。

文章插图

文章插图

6、RPC（远程过程调用）给张图自己体会吧（官网没给示例代码，我也就不写了），就是通过两个交换机实现一个可回调的过程吧。

文章插图

三、RabbitMq的交换机RabbitMq是有一个交换机的概念的，消息(Message)由Client发送，RabbitMQ接收到消息之后通过交换机转发到对应的队列上面。Worker会从队列中获取未被读取的数据处理。这样就可以实现消息的发送者无需知道消息使用者的存在，反之亦然。Direct exchange：直连（路由）交换机，转发消息到routigKey指定的队列
Fanout exchange：扇形交换机，转发消息到所有绑定队列（相当于广播）
Topic exchange：主题交换机，按规则转发消息（很灵活）
Headers exchange：首部交换机

文章插图

前面的简单类型我们都是忽略了交换机的参数的，如该方法：channel.basicPublish("", QUEUE_NAME, null, message.getBytes());就是这个方法的第一个参数，置空说明使用了默认的交换机。
有几种交换类型可用：direct，topic，headers 和fanout 。