消息队列:
组成:
交换器,队列,绑定
作用:异步处理,削峰,服务解耦
交换器
RabbitMQ常见的exchange(交换器)类型:
-
direct–路由键完全匹配才可以
-
fanout–广播
-
topic --主题,模糊匹配路由键
队列
messagequeue:
组成:
-
路由键 routine-key—决定消息发给谁
-
优先级priority–决定消息发送的优先级
-
分发模式deliver-mode–决定消息的发送方式–持久化等
绑定
binding:
依赖
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId><version>2.6.4</version></dependency>
配置文件:
spring:rabbitmq:username: zzypassword: 1234host: 172.24.232.166
注册队列:
@Configuration
public class RabbitConfig {@Beanprotected Queue queue(){Queue queue= new Queue("zzy");return queue;}
}
生产者发送消息
//作为信息的发布者
@SpringBootTest(classes = ApplicationRabbitMq.class)
public class RabbitTest {//Amqp模板类@Autowiredprivate AmqpTemplate amqpTemplate;@Testvoid RabbitTest(){amqpTemplate.convertAndSend("zzy","hello world");System.out.println("success");}
}
消费者消费消息:
@Component //加注解,不然无法解析
public class RabbitMqConsumer {@RabbitListener(queues = "zzy")//订阅的队列public void Listened1(String msg){System.out.println("取出的消息1----"+msg);}@RabbitListener(queues = "zzy")public void Listened2(String msg){System.out.println("取出的消息2----"+msg);}
}
默认使用direct队列,如果要使用广播队列:
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitFanoutConfig {//准备两个队列@Beanprotected Queue queue() {return new Queue("gavin");}@Beanprotected Queue queue2() {return new Queue("zzy");}//内置fanout交换器名称amq.fanout@Beanprotected FanoutExchange fanoutExchange() {return new FanoutExchange("amq.fanout");}//将交换器和队列绑定@Beanprotected Binding fanoutBinding(Queue queue, FanoutExchange fanoutExchange) {return BindingBuilder.bind(queue).to(fanoutExchange);}@Beanprotected Binding fanoutBinding2(Queue queue2, FanoutExchange fanoutExchange) {return BindingBuilder.bind(queue2).to(fanoutExchange);}
}
一个交换器可以绑定多个队列;
几个重要的注解/Bean
import org.springframework.amqp.core.AmqpTemplate;
@Bean
AmqpTemplate amqpTemplate
//此模板中有发送消息的方法-------------------------------------
import org.springframework.amqp.core.Queue;
//注册一个队列direct@Beanprotected Queue queue() {return new Queue("gavin");}//注册一个广播交换器@Beanprotected FanoutExchange fanoutExchange() {return new FanoutExchange("amq.fanout");}
//交换器绑定队列//将交换器和队列绑定@Beanprotected Binding fanoutBinding(Queue queue, FanoutExchange fanoutExchange) {return BindingBuilder.bind(queue).to(fanoutExchange);}
//注册一个topic交换器@Beanprotected TopicExchange topicExchange() {return new TopicExchange("amq.topic");}
//绑定队列//将交换器和队列绑定@Beanprotected Binding TopicBind1(Queue queue, TopicExchange topicExchange) {return BindingBuilder.bind(queue).to(topicExchange).with("com.gavin.*");//匹配路由规则}
消费者:
@RabbitListener(queues = "gavin")该方法从对列中消费消息
消息重复消费原因:
消费完毕后本该向broker发送ack,但是由于网路延迟较高过了broker等待的时间,于是broker会把消息再次投递到consumer
解决方案:
数据库—处理消息前,使用消息主键在表中带有约束的字段中insert,插入成功则消费成功,插入失败则已经消费过了,不再进行消费
Map–单机版的使用ConrrentHashMap ->putifAbsent
Redis --分布式锁
保证消息队列的消费顺序
同一个topic,同一个queue,发的时候让一个线程去发,消费的时候让一个线程去消费,如果多线程暂时无法保证消费的有序性
怎么保证消息发送到同一个queue?
RocketMQ 提供了一个MessageQueueSelector 接口,重写接口方法
RocketMQ如何保证消息不丢失
Producer端:
采用send()同步发消息,发送结果是同步感知的;
Broker端:
设置数显策略为同步刷新策略
集群部署,配置主从,高可用模式
Consumer端:
消费正常后再进行手动ACK确认
)
