网站公司建设公司,上海网站建设就q479185700顶上,校园网站建设年度总结,wordpress会员推广系统RabbitMQ的5种消息队列
1、七种模式介绍与应用场景
1.1 简单模式(Hello World) 一个生产者对应一个消费者#xff0c;RabbitMQ 相当于一个消息代理#xff0c;负责将 A 的消息转发给 B。
应用场景#xff1a;将发送的电子邮件放到消息队列#xff0c;然后邮件服务在队列…RabbitMQ的5种消息队列
1、七种模式介绍与应用场景
1.1 简单模式(Hello World) 一个生产者对应一个消费者RabbitMQ 相当于一个消息代理负责将 A 的消息转发给 B。
应用场景将发送的电子邮件放到消息队列然后邮件服务在队列中获取邮件并发送给收件人。
1.2 工作队列模式(Work queues) 在多个消费者之间分配任务(竞争的消费者模式)一个生产者对应多个消费者一般适用于执行资源密集型任务
单个消费者处理不过来需要多个消费者进行处理。
应用场景一个订单的处理需要10s有多个订单可以同时放到消息队列然后让多个消费者同时处理这样就是
并行了而不是单个消费者的串行情况。
1.3 订阅模式(Publish/Subscribe) 一次向许多消费者发送消息一个生产者发送的消息会被多个消费者获取也就是将消息将广播到所有的消费者
中。
应用场景更新商品库存后需要通知多个缓存和多个数据库这里的结构应该是
一个 fanout 类型交换机扇出两个个消息队列分别为缓存消息队列、数据库消息队列一个缓存消息队列对应着多个缓存消费者一个数据库消息队列对应着多个数据库消费者
1.4 路由模式(Routing) 有选择地 (Routing key) 接收消息发送消息到交换机并且要指定路由 key消费者将队列绑定到交换机时需要指
定路由 key仅消费指定路由 key 的消息。
应用场景如在商品库存中增加了 1 台 iphone12iphone12 促销活动消费者指定 routing key 为 iphone12只
有此促销活动会接收到消息其它促销活动不关心也不会消费此 routing key 的消息。
1.5 主题模式(Topics) 根据主题 (Topics) 来接收消息将路由 key 和某模式进行匹配此时队列需要绑定在一个模式上#匹配一个词
或多个词*只匹配一个词。
应用场景同上 iphone 促销活动可以接收主题为 iphone 的消息如 iphone12、iphone13 等。
1.6 远程过程调用(RPC) 如果我们需要在远程计算机上运行功能并等待结果就可以使用 RPC具体流程可以看图。
应用场景需要等待接口返回数据如订单支付。
1.7 发布者确认(Publisher Confirms)
与发布者进行可靠的发布确认发布者确认是 RabbitMQ 扩展可以实现可靠的发布。在通道上启用发布者确认
后RabbitMQ 将异步确认发送者发布的消息这意味着它们已在服务器端处理。
应用场景对于消息可靠性要求较高比如钱包扣款。
2、代码演示
dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.6.0/version
/dependency2.1 简单模式
package simple;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Sender {private final static String QUEUE_NAME simple_queue;public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);factory.setPort(5672);Connection connection factory.newConnection();Channel channel connection.createChannel();// 声明队列// queue: 队列名// durable: 是否持久化// exclusive: 是否排外,即只允许该channel访问该队列,一般等于true的话用于一个队列只能有一个消费者来消费的场景// autoDelete: 是否自动删除,消费完删除// arguments: 其他属性channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 消息内容String message simplest mode message;channel.basicPublish(, QUEUE_NAME, null, message.getBytes());System.out.println([x]Sent message );// 最后关闭通关和连接channel.close();connection.close();}
}package simple;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Receiver {private final static String QUEUE_NAME simple_queue;public static void main(String[] args) throws IOException, TimeoutException {// 获取连接ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);factory.setPort(5672);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( [x] Received delivery.getEnvelope().getRoutingKey() : message );};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag - {});}
}[x]Sent simplest mode message[x] Received simple_queue:simplest mode message2.2 工作队列模式
package work;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Receiver1 {private final static String QUEUE_NAME queue_work;public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);factory.setPort(5672);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 同一时刻服务器只会发送一条消息给消费者channel.basicQos(1);DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( [x-1] Received delivery.getEnvelope().getRoutingKey() : message );};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag - {});}
}package work;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Receiver2 {private final static String QUEUE_NAME queue_work;public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);factory.setPort(5672);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 同一时刻服务器只会发送一条消息给消费者channel.basicQos(1);DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( [x-2] Received delivery.getEnvelope().getRoutingKey() : message );};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag - {});}
}package work;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Sender {private final static String QUEUE_NAME queue_work;public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);factory.setPort(5672);Connection connection factory.newConnection();Channel channel connection.createChannel();// 声明队列channel.queueDeclare(QUEUE_NAME, false, false, false, null);for (int i 0; i 10; i) {String message work mode message i;channel.basicPublish(, QUEUE_NAME, null, message.getBytes());System.out.println([x] Sent message );Thread.sleep(i * 10);}channel.close();connection.close();}
}[x] Sent work mode message0
[x] Sent work mode message1
[x] Sent work mode message2
[x] Sent work mode message3
[x] Sent work mode message4
[x] Sent work mode message5
[x] Sent work mode message6
[x] Sent work mode message7
[x] Sent work mode message8
[x] Sent work mode message9[x-1] Received queue_work:work mode message0
[x-1] Received queue_work:work mode message2
[x-1] Received queue_work:work mode message4
[x-1] Received queue_work:work mode message6
[x-1] Received queue_work:work mode message8[x-2] Received queue_work:work mode message1
[x-2] Received queue_work:work mode message3
[x-2] Received queue_work:work mode message5
[x-2] Received queue_work:work mode message7
[x-2] Received queue_work:work mode message92.3 发布订阅模式
package publishsubscribe;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;public class Receive1 {private static final String EXCHANGE_NAME logs;public static void main(String[] argv) throws Exception {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, fanout);String queueName channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, );System.out.println( [*] Waiting for messages. To exit press CTRLC);// 订阅消息的回调函数DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( [x-1] Received message );};// 消费者有消息时出发订阅回调函数channel.basicConsume(queueName, true, deliverCallback, consumerTag - {});}
}package publishsubscribe;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;public class Receive2 {private static final String EXCHANGE_NAME logs;public static void main(String[] argv) throws Exception {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, fanout);String queueName channel.queueDeclare().getQueue();channel.queueBind(queueName, EXCHANGE_NAME, );System.out.println( [*] Waiting for messages. To exit press CTRLC);// 订阅消息的回调函数DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( [x-2] Received2 message );};// 消费者有消息时出发订阅回调函数channel.basicConsume(queueName, true, deliverCallback, consumerTag - {});}
}package publishsubscribe;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Sender {private static final String EXCHANGE_NAME logs;public static void main(String[] argv) throws Exception {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, fanout);String message publish subscribe message;channel.basicPublish(EXCHANGE_NAME, , null, message.getBytes(UTF-8));System.out.println( [x] Sent message );channel.close();connection.close();}
}[x] Sent publish subscribe message[*] Waiting for messages. To exit press CTRLC
[x-1] Received publish subscribe message[*] Waiting for messages. To exit press CTRLC
[x-2] Received2 publish subscribe message2.4 路由模式
package router;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Receiver1 {private final static String QUEUE_NAME queue_routing;private final static String EXCHANGE_NAME exchange_direct;public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);factory.setPort(5672);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 指定路由的key,接收key和key2channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, key);channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, key2);channel.basicQos(1);DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( [x-1] Received delivery.getEnvelope().getRoutingKey() : message );};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag - {});}
}package router;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Receiver2 {private final static String QUEUE_NAME queue_routing2;private final static String EXCHANGE_NAME exchange_direct;public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);factory.setPort(5672);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 仅接收key2channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, key2);channel.basicQos(1);DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( [x-2] Received delivery.getEnvelope().getRoutingKey() : message );};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag - {});}
}package router;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Sender {private final static String EXCHANGE_NAME exchange_direct;private final static String EXCHANGE_TYPE direct;public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);factory.setPort(5672);Connection connection factory.newConnection();Channel channel connection.createChannel();// 交换机声明channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);// 只有routingKey相同的才会消费String message routing mode message;channel.basicPublish(EXCHANGE_NAME, key2, null, message.getBytes());System.out.println([x] Sent message );// channel.basicPublish(EXCHANGE_NAME, key, null, message.getBytes());// System.out.println([x] Sent message );channel.close();connection.close();}
}[x] Sent routing mode message[x-1] Received key2:routing mode message[x-2] Received key2:routing mode message2.5 主题模式
package topic;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Receiver1 {private final static String QUEUE_NAME queue_topic;private final static String EXCHANGE_NAME exchange_topic;public static void main(String[] args) throws IOException, InterruptedException, TimeoutException {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);factory.setPort(5672);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);// 可以接收key.1channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, key.*);channel.basicQos(1);DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( [x-1] Received delivery.getEnvelope().getRoutingKey() : message );};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag - {});}
}package topic;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;import java.io.IOException;
import java.util.concurrent.TimeoutException;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, TimeoutException {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);factory.setPort(5672);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(QUEUE_NAME, false, false, false, null);// *号代表单个单词可以接收key.1channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, *.*);// #号代表多个单词可以接收key.1.2channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, *.#);channel.basicQos(1);DeliverCallback deliverCallback (consumerTag, delivery) - {String message new String(delivery.getBody(), UTF-8);System.out.println( [x-2] Received delivery.getEnvelope().getRoutingKey() : message );};channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag - {});}
}package topic;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Sender {private final static String EXCHANGE_NAME exchange_topic;private final static String EXCHANGE_TYPE topic;public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory new ConnectionFactory();factory.setHost(localhost);factory.setPort(5672);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.exchangeDeclare(EXCHANGE_NAME, EXCHANGE_TYPE);String message topics model message with key.1;channel.basicPublish(EXCHANGE_NAME, key.1, null, message.getBytes());System.out.println([x] Sent message );String message2 topics model message with key.1.2;channel.basicPublish(EXCHANGE_NAME, key.1.2, null, message2.getBytes());System.out.println([x] Sent message2 );channel.close();connection.close();}
}[x] Sent topics model message with key.1
[x] Sent topics model message with key.1.2[x-1] Received key.1:topics model message with key.1[x-2] Received key.1:topics model message with key.1
[x-2] Received key.1.2:topics model message with key.1.23、四种交换机介绍 直连交换机( Direct exchange )具有路由功能的交换机绑定到此交换机的时候需要指定一个 routing_key交换机发送消息的时候需要 routing_key会将消息发送道对应的队列。 扇形交换机( Fanout exchange )广播消息到所有队列没有任何处理速度最快。 主题交换机( Topic exchange )在直连交换机基础上增加模式匹配也就是对 routing_key 进行模式匹配 * 代表一个单词# 代表多个单词。 首部交换机( Headers exchange )忽略 routing_key使用 Headers 信息(一个 Hash 的数据结构)进行匹 配优势在于可以有更多更灵活的匹配规则。
