RabbitMQ五种工作模式

客官°小女子只卖身不卖艺 2022-10-30 01:25 244阅读 0赞

# RabbitMQ五种工作模式 #

### RabbitMQ五种工作模式 ###

*在SpringBoot环境下做的代码测试，RabbitMQ的包是用SpringBoot的starter-amqp包引入的。*

#### 1、简单队列 ####

![在这里插入图片描述][20210216213718429.png_pic_center]

一个生产者对应一个消费者！！！

1、pom文件

SpringBoot导入rabbitmq 启动包

<dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-amqp</artifactId>
 </dependency>

2、工具类

/** * 〈简述〉 * 〈连接RabbitMQ的工具类〉 * * @create 2020/7/1 * @since 1.0.0 */
 public class ConnectionUtil { 
 public static Connection getConnection() throws Exception { 
 return getConnection(new Properties());
 }
 
 private static Connection getConnection(Properties properties) throws Exception { 
 return getConnection(properties.getHost(),
 properties.getPort(),
 properties.getvHost(),
 properties.getUserName(),
 properties.getPassWord());
 }
 
 public static Connection getConnection(String host, int port, String vHost, String userName, String passWord) throws Exception { 
 //1、定义连接工厂
 ConnectionFactory factory = new ConnectionFactory();
 //2、设置服务器地址
 factory.setHost(host);
 //3、设置端口
 factory.setPort(port);
 //4、设置虚拟主机、用户名、密码
 factory.setVirtualHost(vHost);
 factory.setUsername(userName);
 factory.setPassword(passWord);
 //5、通过连接工厂获取连接
 Connection connection = factory.newConnection();
 return connection;
 }
 
 public static class Properties implements Serializable { 
 String host = "192.168.1.103";
 int port = 5672;
 String vHost = "/";
 String userName = "guest";
 String passWord = "guest";
 
 public Properties() { 
 }
 
 public Properties(String host, int port, String vHost, String userName, String passWord) { 
 this.host = host;
 this.port = port;
 this.vHost = vHost;
 this.userName = userName;
 this.passWord = passWord;
 }
 
 public String getHost() { 
 return host;
 }
 
 public Properties setHost(String host) { 
 this.host = host;
 return self();
 }
 
 public int getPort() { 
 return port;
 }
 
 public Properties setPort(int port) { 
 this.port = port;
 return self();
 }
 
 public String getvHost() { 
 return vHost;
 }
 
 public Properties setvHost(String vHost) { 
 this.vHost = vHost;
 return self();
 }
 
 public String getUserName() { 
 return userName;
 }
 
 public Properties setUserName(String userName) { 
 this.userName = userName;
 return self();
 }
 
 public String getPassWord() { 
 return passWord;
 }
 
 public Properties setPassWord(String passWord) { 
 this.passWord = passWord;
 return self();
 }
 
 private Properties self(){ 
 return this;
 }
 }
 }

3、生产者 Producer

/** * 〈简述〉 * 〈简单队列——消息生产者〉 * * @create 2020/7/1 * @since 1.0.0 */
 public class Producer { 
 private final static String QUEUE_NAME = QueueName.test_simple_queue.toString();
 
 public static void main(String[] args) throws Exception { 
 sendMessage();
 }
 
 public static void sendMessage() throws Exception { 
 //1、获取连接
 Connection connection = ConnectionUtil.getConnection();
 //2、声明信道
 Channel channel = connection.createChannel();
 //3、声明(创建)队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 //4、定义消息内容
 String message = "hello rabbitmq ";
 //5、发布消息
 channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
 System.out.println("[x] Sent'" + message + "'");
 //6、关闭通道
 channel.close();
 //7、关闭连接
 connection.close();
 }
 }

4、消费者Consumer

/** * 〈简述〉 * 〈消息消费者〉 * * @create 2020/7/1 * @since 1.0.0 */
 public class Customer { 
 
 private final static String QUEUE_NAME = QueueName.test_simple_queue.toString();
 
 public static void main(String[] args) throws Exception { 
 getMessage();
 
 }
 
 public static void getMessage() throws Exception { 
 //1、获取连接
 Connection connection = ConnectionUtil.getConnection();
 //2、声明通道
 Channel channel = connection.createChannel();
 //3、声明队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 //4、定义队列的消费者
 Consumer consumer = new DefaultConsumer(channel) { 
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope,
 AMQP.BasicProperties properties, byte[] body) throws IOException { 
 String msgString = new String(body, "utf-8");
 System.out.println("接收的消息：" + msgString);
 }
 };
 //5、监听队列
 /* true:表示自动确认，只要消息从队列中获取，无论消费者获取到消息后是否成功消费，都会认为消息已经成功消费 false:表示手动确认，消费者获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈， 如果消费者一直没有反馈，那么该消息将一直处于不可用状态，并且服务器会认为该消费者已经挂掉，不会再给其 发送消息，直到该消费者反馈。 */
 channel.basicConsume(QUEUE_NAME, true, consumer);
 }
 }

> 注意这里消费者有自动确认消息和手动确认消息两种模式。

#### 2、work 模式 ####

![在这里插入图片描述][20210216213754865.png_pic_center]

一个生产者对应多个消费者，但是一条消息只能有一个消费者获得消息！！！

2.1轮询分发  
　  
　　1、生产者

/** * 〈简述〉 * 〈轮询分发——生产者〉 * * @create 2020/7/3 * @since 1.0.0 */
 public class Send { 
 private static final String QUEUE_NAME = QueueName.test_work_queue.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 for (int i = 0; i < 50; i++) { 
 String msg = "hello " + i;
 System.out.println("[mq] send:" + msg);
 channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
 Thread.sleep(i * 20);
 }
 channel.close();
 connection.close();
 }
 
 }

2、消费者

这里创建两个消费者

消费者1：每接收一条消息后休眠1秒

/** * 〈简述〉 * 〈接收者〉 * * @create 2020/7/3 * @since 1.0.0 */
 public class Receive1 { 
 private static final String QUEUE_NAME = QueueName.test_work_queue.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel、
 Channel channel = connection.createChannel();
 //声明队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 //定义一个消费这
 Consumer consumer = new DefaultConsumer(channel) { 
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
 String msg = new String(body, "utf-8");
 System.out.println("[1] Receive1 msg:" + msg);
 try { 
 Thread.sleep(1000);
 } catch (InterruptedException e) { 
 e.printStackTrace();
 } finally { 
 System.out.println("[1] done");
 }
 }
 };
 boolean autoAck = false;
 channel.basicConsume(QUEUE_NAME, autoAck, consumer);
 }
 
 }

消费者2：每接收一条消息后休眠2秒

/** * 〈简述〉 * 〈接收者〉 * * @create 2020/7/3 * @since 1.0.0 */
 public class Receive2 { 
 private static final String QUEUE_NAME = QueueName.test_work_queue.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 //定义一个消费这
 Consumer consumer = new DefaultConsumer(channel) { 
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
 String msg = new String(body, "utf-8");
 System.out.println("[2] Receive2 msg:" + msg);
 try { 
 Thread.sleep(2000);
 } catch (InterruptedException e) { 
 e.printStackTrace();
 } finally { 
 System.out.println("[2] done");
 }
 }
 };
 boolean autoAck = false;
 channel.basicConsume(QUEUE_NAME, autoAck, consumer);
 }
 
 }

3、测试结果

生产者从0-49依次发送消息 
    	消费者1：接收到偶数 	
    	消费者2：接收到奇数

4、结论

*轮询分发就是将消息队列中的消息，依次发送给所有消费者。一个消息只能被一个消费者获取。*

2.2公平分发  
消费者关闭自动应答，开启手动回执

/** * 〈简述〉 * 〈接收者〉 * * @create 2020/7/3 * @since 1.0.0 */
 public class Receive2 { 
 private static final String QUEUE_NAME = QueueName.test_work_queue.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 channel.basicQos(1);//保证一次只分发一个消息
 //定义一个消费这
 Consumer consumer = new DefaultConsumer(channel) { 
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
 String msg = new String(body, "utf-8");
 System.out.println("[2] Receive2 msg:" + msg);
 try { 
 Thread.sleep(2000);
 } catch (InterruptedException e) { 
 e.printStackTrace();
 } finally { 
 System.out.println("[2] done");
 //手动回执
 channel.basicAck(envelope.getDeliveryTag(),false);
 }
 }
 };
 boolean autoAck = false;//自动应答
 channel.basicConsume(QUEUE_NAME, autoAck, consumer);
 }
 
 }

*手动回执：消费者完成业务接口方法后可以告知消息队列处理完成，消息队列从队列中取一条消息发送给消费者。*  
　　*能者多劳：效率高的消费者消费消息多。*

#### 3、发布/订阅模式 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2lUX01hTm9uZ2tpbmc_size_16_color_FFFFFF_t_70_pic_center]

一个消费者将消息首先发送到交换器，交换器绑定到多个队列，然后被监听该队列的消费者所接收并消费。

*ps:X表示交换器，在RabbitMQ中，交换器主要有四种类型:direct、fanout、topic、headers，这里的交换器是 fanout。下面我们会详细介绍这几种交换器。*

1、生产者

/** * 〈简述〉 * 〈订阅模式——生产者〉 * * @create 2020/7/3 * @since 1.0.0 */
 public class Send { 
 private static final String EXCHANGE_NAME = MqName.exchange_fanout.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明交换机
 channel.exchangeDeclare(EXCHANGE_NAME,"fanout");//分发
 //发送消息
 String msg = "hello exchange";
 System.out.println("[mq] send:" + msg);
 channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());
 channel.close();
 connection.close();
 }
 
 }

2、消费者

注意：两个消费者绑定不同的队列，绑定相同的交换机；

消费者1：绑定队列名=queue\_fanout\_email1

/** * 〈简述〉 * 〈接收者〉 * * @create 2020/7/3 * @since 1.0.0 */
 public class Receive1 { 
 private static final String QUEUE_NAME = MqName.queue_fanout_email.toString() + "1";
 private static final String EXCHANGE_NAME = MqName.exchange_fanout.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 //绑定到交换机
 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
 //定义一个消费这
 Consumer consumer = new DefaultConsumer(channel) { 
 /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
 String msg = new String(body, "utf-8");
 System.out.println("[1] Receive1 msg:" + msg);
 try { 
 Thread.sleep(1000);
 } catch (InterruptedException e) { 
 e.printStackTrace();
 } finally { 
 System.out.println("[1] done");
 }
 }
 };
 boolean autoAck = false;
 channel.basicConsume(QUEUE_NAME, autoAck, consumer);
 }
 
 }

消费者2：绑定队列名=queue\_fanout\_email2

/** * 〈简述〉 * 〈接收者〉 * * @create 2020/7/3 * @since 1.0.0 */
 public class Receive2 { 
 private static final String QUEUE_NAME = MqName.queue_fanout_email.toString() + "2";
 private static final String EXCHANGE_NAME = MqName.exchange_fanout.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 //绑定到交换机
 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
 //定义一个消费这
 Consumer consumer = new DefaultConsumer(channel) { 
 /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
 String msg = new String(body, "utf-8");
 System.out.println("[2] Receive2 msg:" + msg);
 try { 
 Thread.sleep(2000);
 } catch (InterruptedException e) { 
 e.printStackTrace();
 } finally { 
 System.out.println("[2] done");
 }
 }
 };
 boolean autoAck = false;
 channel.basicConsume(QUEUE_NAME, autoAck, consumer);
 }
 
 }

3、测试结果  
*如上图，两个消费者获得了同一条消息。即就是，一个消息从交换机同时发送给了两个队列中，监听这两个队列的消费者消费了这个消息；  
如果没有队列绑定交换机，则消息将丢失。因为交换机没有存储能力，消息只能存储在队列中。*

#### 4、路由模式 ####

![在这里插入图片描述][20210216213900903.png_pic_center]

生产者将消息发送到direct交换器，在绑定队列和交换器的时候有一个路由key，生产者发送的消息会指定一个路由key，那么消息只会发送到相应key相同的队列，接着监听该队列的消费者消费消息。

也就是让消费者有选择性的接收消息。

1、生产者

/** * 〈简述〉 * 〈路由模式-消息发送者〉 * * @create 2020/7/8 * @since 1.0.0 */
 public class Send { 
 
 public static final String EXCHANGE_NAME = MqName.exchange_routing.toString();
 public static final String ROUTING_KEY = MqName.routing_world.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明交换机
 channel.exchangeDeclare(EXCHANGE_NAME, "direct");
 String msg = "route message ->" + ROUTING_KEY;
 System.out.println("对 " + ROUTING_KEY + " 发送消息：" + msg);
 channel.basicPublish(EXCHANGE_NAME, ROUTING_KEY, null, msg.getBytes());
 //关闭连接
 channel.close();
 connection.close();
 }
 
 }

2、消费者

注意：两个消费者，绑定相同的交换机，不同的队列，不一样的路由

消费者1：路由=routing\_hello

/** * 〈简述〉 * 〈接收消息1〉 * * @create 2020/7/8 * @since 1.0.0 */
 public class Receive1 { 
 
 public static final String QUEUE_NAME = MqName.queue_routing_001.toString();
 public static final String EXCHANGE_NAME = MqName.exchange_routing.toString();
 public static final String ROUTING_KEY_hello = MqName.routing_hello.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 //设置预读取数
 channel.basicQos(1);
 //绑定交换机
 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_hello);
 Consumer consumer = new DefaultConsumer(channel) { 
 /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
 String msg = new String(body, "utf-8");
 System.out.println(envelope.getRoutingKey() + " [1] Receive1 msg:" + msg);
 }
 };
 channel.basicConsume(QUEUE_NAME, true, consumer);
 }
 
 }

消费者2：路由1=routing\_world 路由2=routing\_hello

/** * 〈简述〉 * 〈接收消息2〉 * * @create 2020/7/8 * @since 1.0.0 */
 public class Receive2 { 
 
 public static final String QUEUE_NAME = MqName.queue_routing_002.toString();
 public static final String EXCHANGE_NAME = MqName.exchange_routing.toString();
 public static final String ROUTING_KEY_world = MqName.routing_world.toString();
 public static final String ROUTING_KEY_hello = MqName.routing_hello.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 //设置预读取数
 channel.basicQos(1);
 //绑定交换机和路由器，可以绑定多个路由
 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_world);
 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY_hello);
 //定义消息消费者
 Consumer consumer = new DefaultConsumer(channel) { 
 /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
 String msg = new String(body, "utf-8");
 System.out.println(envelope.getRoutingKey() + " [2] Receive1 msg:" + msg);
 }
 };
 //接收消息
 channel.basicConsume(QUEUE_NAME, true, consumer);
 }
 
 }

3、测试结果

> 生产者发送：routing\_world
> 
>  消费者1：没有接收到
> 
>  消费者2：接收到了
> 
>  生产者发送：routing\_hello
> 
>  消费者1：接收到了
> 
>  消费者2：接收到了

*路由模式，是以路由规则为导向，引导消息存入符合规则的队列中。再由队列的消费者进行消费的。*

#### 5、主题模式 ####

![在这里插入图片描述][20210216213927538.png_pic_center]

上面的路由模式是根据路由key进行完整的匹配（完全相等才发送消息），这里的通配符模式通俗的来讲就是模糊匹配。

符号“\#”表示匹配一个或多个词，符号“\*”表示匹配一个词。

1、生产者

/** * 〈简述〉 * 〈主题模式-消息发送者〉 * * @create 2020/7/8 * @since 1.0.0 */
 public class Send { 
 
 public static final String EXCHANGE_NAME = MqName.exchange_topic.toString();
 public static final String ROUTING_KEY = MqName.routing_goods.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明交换机
 channel.exchangeDeclare(EXCHANGE_NAME, "topic");
 // String routingKey = ROUTING_KEY + ".add";
 String routingKey = ROUTING_KEY + ".publish";
 // String routingKey = ROUTING_KEY + ".update";
 
 String msg = "route message ->" + routingKey;
 channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());
 System.out.println("对 " + routingKey + " 发送消息：" + msg);
 //关闭连接
 channel.close();
 connection.close();
 }
 
 }

2、消费者

注意两个消费者，路由的不同

消费者1：路由1=routing\_goods.add 路由2=routing\_goods.update

/** * 〈简述〉 * 〈接收消息1〉 * * @create 2020/7/8 * @since 1.0.0 */
 public class Receive1 { 
 
 public static final String QUEUE_NAME = MqName.queue_topic_001.toString();
 public static final String EXCHANGE_NAME = MqName.exchange_topic.toString();
 public static final String ROUTING_KEY = MqName.routing_goods.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 //设置预读取数
 channel.basicQos(1);
 //绑定交换机
 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY + ".add");
 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY + ".update");
 Consumer consumer = new DefaultConsumer(channel) { 
 /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
 String msg = new String(body, "utf-8");
 System.out.println(envelope.getRoutingKey() + " [1] Receive1 msg:" + msg);
 }
 };
 channel.basicConsume(QUEUE_NAME, true, consumer);
 }
 
 }

消费者2：路由=routing\_goods.\*

/** * 〈简述〉 * 〈接收消息2〉 * * @create 2020/7/8 * @since 1.0.0 */
 public class Receive2 { 
 
 public static final String QUEUE_NAME = MqName.queue_routing_002.toString();
 public static final String EXCHANGE_NAME = MqName.exchange_topic.toString();
 public static final String ROUTING_KEY = MqName.routing_goods.toString();
 
 public static void main(String args[]) throws Exception { 
 //获取连接
 Connection connection = ConnectionUtil.getConnection();
 //获取channel
 Channel channel = connection.createChannel();
 //声明队列
 channel.queueDeclare(QUEUE_NAME, false, false, false, null);
 //设置预读取数
 channel.basicQos(1);
 //绑定交换机和路由器，可以绑定多个路由
 channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY + ".*");
 //定义消息消费者
 Consumer consumer = new DefaultConsumer(channel) { 
 /** * No-op implementation of {@link Consumer#handleDelivery}. * * @param consumerTag * @param envelope * @param properties * @param body */
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
 String msg = new String(body, "utf-8");
 System.out.println(envelope.getRoutingKey() + " [2] Receive1 msg:" + msg);
 }
 };
 //接收消息
 channel.basicConsume(QUEUE_NAME, true, consumer);
 }
 
 }

3、测试结果

> 消费者1只能接收到`.add` 和 `.update`的消息
> 
>  消费者2可以接收到.`add .publish` 和 `.update`的消息
> 
>  与路由模式相似，但是，主题模式是一种模糊的匹配方式。

#### 6.工作模式总结 ####

> ***这五种工作模式，可以归为三类： 生产者，消息队列，一个消费者； 生产者，消息队列，多个消费者；  
> 生产者，交换机，多个消息队列，多个消费者；***

#### 7、四种交换器 ####

> **1、direct 如果路由键完全匹配的话，消息才会被投放到相应的队列。 　  
> 　　2、fanout 当发送一条消息到fanout交换器上时，它会把消息投放到所有附加在此交换器上的队列。  
> 　　3、topic　设置模糊的绑定方式，“\*”操作符将“.”视为分隔符，匹配单个字符；“\#”操作符没有分块的概念，它将任意“.”均视为关键字的匹配部分，能够匹配多个字符。  
>  4、header headers 交换器允许匹配 AMQP 消息的 header 而非路由键，除此之外，header 交换器和  
> direct 交换器完全一致，但是性能却差很多，因此基本上不会用到该交换器**

\*\*

[20210216213718429.png_pic_center]: /images/20221024/605ee2c8c05e4250babfc035a55062fa.png
[20210216213754865.png_pic_center]: /images/20221024/87c372de2169412dabb4faa1745b61cd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2lUX01hTm9uZ2tpbmc_size_16_color_FFFFFF_t_70_pic_center]: /images/20221024/42b9f1d2cc9c47a5a2983276061cd710.png
[20210216213900903.png_pic_center]: /images/20221024/672b7c0807cb469d9feee3d4775b2954.png
[20210216213927538.png_pic_center]: /images/20221024/04dc8f6146884a62ae500282c2a8ff57.png