RabbitMQ的五种⼯作模式
RabbitMQ的五种⼯作模式
RabbitMQ的作⽤
1、服务解藕：降低服务之间耦合度
2、流量削峰：当⾼峰情况下，出现⼤量的请求时，将请求发送到消息队列服务器上，排队等待被处理 3、异步调⽤：将数据发送到消息队列服务器中，系统可以得到⽴即响应，整条链路的响应时间只有200毫秒左右
所需的model类
public interface Constants {
String Sub_Teacher="校长";
String SUB_EXCHANGE = "yuque52";
基因敲除小鼠String SIMPLE_QUEUE = "query52";
String ROUTER_RIBBIT="路由模式";
String TOPIC_RIBBIT="主题模式";
}
public class ChannelUtil {
public static Channel channel(){
//建⽴连接
try {
ConnectionFactory cf = new ConnectionFactory();
cf.setHost("localhost");
cf.setPort(5672);
cf.setUsername("guest");
cf.setPassword("guest");
cf.setVirtualHost("/");
Connection connection = null;
connection = cf.newConnection();
//建⽴信道
Channel channel = ateChannel();
return channel;
} catch (IOException e) {
e.printStackTrace();
} catch (TimeoutException e) {
e.printStackTrace();
}
return null;
}
}
⼀：简单模式（点对点模式）
原理：⽣产者将消息交给默认的交换机，交换机获取消息后交给绑定这个⽣产者的队列（投递规则为队列名称和routing key 相同的队列），监听当前队列的消费者获取信息并执⾏消费逻辑。 使⽤场景：⽤户通过接收⼿机验证码进⾏注册，页⾯上点击获取验证码后，将验证码放到消息队列，然后短信服务从队列中获取到验证码****，并发送给⽤户、聊天室
⼀个⽣产者投递消息给队列 只能允许有⼀个消费者进⾏消费 如果集的话 会进⾏均摊消费 服务器配置不⼀样 均摊就不优了，⼀个⽣产者投递消息给队列 只能允许有⼀个消费者进⾏消费 ，如果集的话，会进⾏均摊消费 ，服务器配置不⼀样，均摊就不优了 ⽣产者：
public class SimplePublisher {
public static void main(String[] args) throws IOException, TimeoutException {
//创建交换机
//创建消息队列
/*
queue – 队列名
durable – 是否持久化，如果为true则会持久化 (服务器重启后仍存在)
exclusive – 是否独占，true则为独占队列，其他链接不能使⽤
autoDelete – 是否⾃动删除消息，true则在使⽤完后删除
arguments – map类型，其他参数
* */
Channel channel = ChannelUtil.channel();
channel.queueDeclare("queue52",false,false,true,null);
//发送消息
/*
exchange 交换机
routing-key 路由键
properties 属性
body 消息体
*/
Scanner scanner = new Scanner(System.in);
print("请输⼊你要发送的消息");
String msg =();
channel.basicPublish("","queue52",Bytes("UTF-8"));
System.out.println("[x] 消息发送成功！ "+msg);
}
private static void print(String str) {
System.out.println("发布者："+str);
}
}
消费者：
public class SimpleConsumer {
public static void main(String[] args) throws IOException {
聚氨酯改性环氧树脂//建⽴连接
//建⽴信道
Channel channel = ChannelUtil.channel();
//创建交换机
//创建消息队列
//绑定交换机
//接受消息
channel.basicConsume("queue52",
(consumerTag,delivery)->{
//处理消息
byte[] body = Body();
String message = new String(body,"utf-8");
System.out.println("[x] 收到消息"+message);
},(consumerTag)->{});
}
}
⼆：⼯作队列（单发送多接收）
默认是Round-robin
原理：⽣产者将消息交给交换机，交换机交给绑定的队列，队列有多个消费者监听，⼀条消息只能由⼀个消费者消费，这样就形成了资源竞争，谁的资源空闲⼤，争抢到的可能性就⼤。
使⽤场景：即⼯作竞争模式或者⼜称消费者竞争模式。即谁先拿到消息，谁来负责 消费。常见的业务场景有抢红包，秒杀、⼤型项⽬中的资源调度过程(直接由最空闲的系统争抢到资源处理任务) ，⼀个电商平台，有两个订单服务，⽤户下单的时候，任意⼀个订单服务消费⽤户的下单请求⽣成订单即可。不⽤两个订单服务同时消费⽤户的下单请求，分布式的任务派发，为了保证消息发送的可靠性，不丢失，使消息持久化，同时为了防⽌接收端在处理消息时down掉，只有在消息处理完成后才发送ack消息。
即循环调度，轮流所有的消费者 发送消息。当消费者稳定运⾏时，他们接受的消息数量是⼀样的。体现在⼯作任务的派发时，他是"公平派遣"。
⽣产者:
public class WorkPublisher {
public static void main(String[] args)throws IOException, TimeoutException {
/
/建⽴连接和通道
Channel channel = ChannelUtil.channel();
//创建交换机，类型是FANPUT
channel.queueDeclare(Constants.Sub_Teacher, false,false,false,null);
//创建队列到消费者端执⾏
Scanner scanner = new Scanner(System.in);
while (true){
print("请输⼊你要发送的消息");
String ();
if ("q".equals(msg)){
break;
}
//发送给交换机不提供routing-key
channel.basicPublish("",Constants.Sub_Teacher,Bytes("utf-8"));
print("消息发送成功: "+msg);
}
}
private static void print(String str) {
System.out.println("发布者："+str);
}
}
消费者
private String name;
public WorkConsumer(String name) {中国兽药监察所
this.name = name;
}
public void consume() throws IOException {
//建⽴信道和连接
Channel channel = ChannelUtil.channel();
//创建消息队列
//绑定在交换机上
//监听队列中的消息接收消息
channel.basicConsume(
Constants.Sub_Teacher,
悬架设计
true,
(consumerTag,delivery)->{
//处理消息
byte[] body = Body();
String message = new String(body,"utf-8");
System.out.println(name + " 收到消息是："+message);
},(consumerTag)->{});
}
nrm
三：发布订阅模式
原理：⽣产者将消息扔给交换机，交换机类型是fanout，不同的队列注册到交换机上，不同的消费注册在不同的队列上。所有消费者都会收到消息。
场景：有⼀个商城，我们新添加⼀个商品后，可能同时需要去更新缓存和数据库。邮件发，聊天,⼴告
交换机没有保存功能，如果没有消费者，则消息会丢失。
⽣产者：
public static void main(String[] args) throws IOException, TimeoutException {
//建⽴信道和连接
Channel channel = ChannelUtil.channel();
//创建交换机，类型是FANOUT
//创建队列到消费者执⾏
//发送消息
Scanner scanner = new Scanner(System.in);
while (true){
print("请输⼊你要发送的消息");
String ();
if ("q".equals(msg)){
break;
}
//发送给交换机，不提供routing-key
channel.basicPublish(Constants.Sub_Teacher,"",Bytes("utf-8"));
print("消息发送成功："+msg);
}
}
private static void print(String str) {
System.out.println("发布者："+str);
}
}
消费者
public class SubConsumer {
private String name;
public SubConsumer(String name) {
血清铁蛋白
this.name = name;
}
public void consume() throws IOException {
/
/建⽴信道和连接
Channel channel = ChannelUtil.channel();
//创建交换机
//创建消息队列
String ribbitName=channel.queueDeclare().getQueue();
//绑定交换机
channel.queueBind(ribbitName, Constants.Sub_Teacher,"");
//接收消息
channel.basicConsume(ribbitName,true,
(consumerTag,delivery)->{
//处理消息
byte[] body = Body();
String message = new String(body,"utf-8");
System.out.println(name + "[x] 收到消息"+message);
},(consumerTag)->{});
}
}
四：路由模式
原理：⽣产者将消息发送给交换机，消息携带具体的routingkey。交换机类型是direct，接收到消息中的routingkey，⽐对与之绑定的队列的routingkey，分发到不同的队列上。
场景：⼀个商城，新添加了⼀个商品，实时性不是很⾼，只需要添加到数据库即可，不⽤刷新缓存。项⽬中的error报错
豫公网安备41160202000603
站长QQ:729038198
关于我们
投诉建议