一、RabbitMq基础知识
0、概述
音讯队列的作用就是接管音讯生产者的音讯，而后将音讯发送到消费者
1、信道channel
我的了解是生产者/消费者和rabbitmq交互的一个通道，负责交换机、队列治理；音讯公布和生产治理；事务管理等
2、交换机
四种交换机：
direct：能够用一个或者多个key绑定到一个或者多个队列上
topic：反对路由的适配符 # *
Fanout播送：将音讯发送给所有的队列
Header头交换机：自定义通过头音讯属性来定义路由的匹配
3、队列：保留音讯的队列
4、消费者：音讯的接收者
5、生产者：音讯的发送者
二、 应用com.rabbitmq.client.*操作mq
2.1、基本操作
0、环境和依赖
<!-- 环境
* jdk 1.8
* idea
* springboot 2.2.6
-->
<!-- 依赖 这里只导入这个包，其中蕴含了Rabbit client的包-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1、创立连贯和信道
//获取连贯
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");//mq主机地址
factory.setPort(5672);//端口，默认时5672
factory.setUsername("leyou");
factory.setPassword("leyou");
factory.setVirtualHost("/leyou");
Connection connection = factory.newConnection();
//获取信道
Channel channel = connection..createChannel();
2、申明交换机 / 队列 / 绑定交换机和队列
//交换机名，交换机类型
channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);
/**
* 第一个参数是queue：要创立的队列名
* 第二个参数是durable：是否长久化。如果为true，能够在RabbitMQ解体后复原音讯
* 第三个参数是exclusive：true示意一个队列只能被一个消费者占有并生产
* 第四个参数是autoDelete：true示意服务器不在应用这个队列是会主动删除它
* 第五个参数是arguments：包含死信队列，队列的ttl
*/
channel.queueDeclare(QUEUE_ONE,true,false,false,null);
//绑定交换机和队列 队列名，交换机名，routekey
channel.queueBind(QUEUE_ONE,EXCHANGE,GIRL);
3、公布音讯
//1、交换机名 2、routekey 3、mandatory强制(须要return回调时必须设置为true) 4、公布音讯参数 5、音讯
channel.basicPublish(EXCHANGE,GIRL,true,null,"xxx提价了".getBytes());
4、接管音讯
//接管音讯前也须要获取连贯和channel，申明队列
//接管音讯
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
//拿到音讯
System.out.println(new String(body,"utf-8"));
}
};
/**
* 参数阐明
* 1：队列名字
* 2：是否自动应答 autoACk，为false时须要手动ack
* 3：消费者，当接管到消费者时会调用给对象中的 handleDelivery 办法
*/
channel.basicConsume(QUEUE_ONE,true,consumer);
2.2、根本利用
1、性能：
有两个人小明和小华，小明对美女感兴趣，小华对股票和没事感兴趣，应用音讯队列将他们感兴趣的音讯发送给他们两个
2、实现：
（1）写一个类来提供创立连贯和信道；
（2）生产者（发送音讯方）类发送音讯
（3）消费者（接管音讯）类接管音讯
- 连贯类
public class ConnectionUtil {
/**
* 应用原始的rabbitmq client api 操作mq
*/
private static ConnectionFactory factory = new ConnectionFactory();
private static Connection connection;
/*
获取连贯
留神导包：须要导client上面的包
*/
public static Connection getConnection() throws IOException, TimeoutException {
// factory.setHost("localhost");
// factory.setPort(5672);
factory.setUsername("leyou");
factory.setPassword("leyou");
factory.setVirtualHost("/leyou");
connection = factory.newConnection();
return connection;
}
public static void close() throws IOException {
connection.close();
}
/*
创立信道
*/
public static Channel getChannel() throws IOException, TimeoutException {
return getConnection().createChannel();
}
}
- 生产者
//生产者
public class provice{
public void producerMsg() throws IOException, TimeoutException, InterruptedException {
Channel channel = ConnectionUtil.getChannel();
String EXCHANGE = "direct_exchange";
channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);
//定义两个队列名
String QUEUE_ONE = "beauty_queue";
String QUEUE_TWO = "food_queue";
channel.queueDeclare(QUEUE_ONE,true,false,false,null);
channel.queueDeclare(QUEUE_TWO,true,false,false,null);
//定义三个key
String GIRL = "girl";
String SHARE = "share";
String FOOD = "food";
//绑定
channel.queueBind(QUEUE_ONE,EXCHANGE,GIRL);
channel.queueBind(QUEUE_TWO,EXCHANGE,SHARE);
channel.queueBind(QUEUE_TWO,EXCHANGE,FOOD);
//发送音讯
/**
* 参数：1交换机，2routekey 3 mandatory：强制；(须要return回调时必须设置为true)
* 3参数，4音讯字节数据
*/
channel.basicPublish(EXCHANGE,GIRL,true,null,"快看，是她".getBytes());
channel.basicPublish(EXCHANGE,SHARE,true,null,"股票涨了".getBytes());
channel.basicPublish(EXCHANGE,FOOD,true,null,"肯德基提价了".getBytes());
//敞开连贯
channel.close();
ConnectionUtil.close();
}
}
- 消费者
public class ConsumerMq { // 生产音讯
/**
* 应用原始的rabbitmq client api 操作mq
*/
String EXCHANGE = "direct_exchange";
String QUEUE_ONE = "beauty_queue";
String QUEUE_TWO = "food_queue";
//key
String GIRL = "girl";
String SHARE = "share";
String FOOD = "food";
public void consumer() throws IOException, TimeoutException {
Channel channel = ConnectionUtil.getChannel();
/**
* 第一个参数是queue：要创立的队列名
* 第二个参数是durable：是否长久化。如果为true，能够在RabbitMQ解体后复原音讯
* 第三个参数是exclusive：true示意一个队列只能被一个消费者占有并生产
* 第四个参数是autoDelete：true示意服务器不在应用这个队列是会主动删除它
* 第五个参数是arguments：包含死信队列，队列的ttl，
*/
channel.queueDeclare(QUEUE_ONE,true,false,false,null);
channel.queueDeclare(QUEUE_TWO,true,false,false,null);
//在生产者绑定了交换机和队列，在这里就不须要绑定
//channel.queueBind(QUEUE_ONE,EXCHANGE,GIRL);
//channel.queueBind(QUEUE_TWO,EXCHANGE,SHARE);
//channel.queueBind(QUEUE_TWO,EXCHANGE,FOOD);
//接管音讯
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body,"utf-8"));
//手动应答ack能够在该办法中进行；参数：1.音讯tag，2.是否批量ack
channel.basicAck(envelope.getDeliveryTag(),false);
}
};
/**
* 参数阐明
* 1：队列名字
* 2：是否自动应答 autoACk 为false时须要手动ack
* 3：消费者，当接管到消费者时会调用给对象中的 handleDelivery 办法
*/
channel.basicConsume(QUEUE_ONE,false,consumer);
channel.basicConsume(QUEUE_TWO,false,consumer);
}
}
2.3、mq事务，发送方确认，和音讯回调
概述
音讯的发送链路 生产者 -> exchange –> queue –> 消费者；为确保音讯发送到rabbitmq，amqp协定提供了三个机制来保障：事务，发送方确认（ack），音讯回调（returncallback）；事务的形式和数据库的事务相似，这里不做具体介绍；发送方确认是当音讯发送到交换机时， broker（实现amqp协定的服务端，这里指rabbitmq）会回调发送者的一个固定办法来确认音讯胜利发送；音讯回调是产生在交换机通过路由key转发到队列的过程中，如果音讯不能通过key找到对应的queue则回调一个固定办法将音讯返回给生产者，确保音讯不失落
1、mq事务
- rabbitMq是反对事务的，然而应用事务的效率很低，在音讯数量很大的状况下影响性能
2、发送方确认
对于固定音讯体大小和线程数，如果音讯长久化，生产者confirm(或者采纳事务机制)，消费者ack那么对性能有很大的影响.
音讯长久化的优化没有太好办法，用更好的物理存储（SAS, SSD, RAID卡）总会带来改善。生产者confirm这一环节的优化则次要在于客户端程序的优化之上。归纳起来，客户端实现生产者confirm有三种编程形式：
- 一般confirm模式：每发送一条音讯后，调用waitForConfirms()办法，期待服务器端confirm。实际上是一种串行confirm了。
- 批量confirm模式：每发送一批音讯后，调用waitForConfirms()办法，期待服务器端confirm。
- 异步confirm模式：提供一个回调办法，服务端confirm了一条或者多条音讯后Client端会回调这个办法。
- [ ] 一般confirm模式
//要点
//第1种
//一般confirm模式最简略，publish一条音讯后，期待服务器端confirm,如果服务端返回false或者超时工夫内未返回，客户端进行音讯重传。
//1.发消息前
channel.confirmSelect();
//2.发消息后
//判断音讯发送是否胜利
if(channel.waitForConfirms()){
System.out.println("音讯发送胜利");
}
- [ ] 批量confirm模式
批量confirm模式略微简单一点，客户端程序须要定期（每隔多少秒）或者定量（达到多少条）或者两则联合起来publish音讯，而后期待服务器端confirm, 相比一般confirm模式，批量极大晋升confirm效率，然而问题在于一旦呈现confirm返回false或者超时的状况时，客户端须要将这一批次的音讯全部重发，这会带来显著的反复音讯数量，并且，当音讯常常失落时，批量confirm性能应该是不升反降的。
channel.confirmSelect();
for(int i=0;i<batchCount;i++){
channel.basicPublish(ConfirmConfig.exchangeName, ConfirmConfig.routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, ConfirmConfig.msg_10B.getBytes());
}
if(!channel.waitForConfirms()){
System.out.println("send message failed.");
}
异步confirm模式
异步confirm模式的编程实现最简单，Channel对象提供的ConfirmListener()回调办法只蕴含deliveryTag（以后Chanel收回的音讯序号），咱们须要本人为每一个Channel保护一个unconfirm的音讯序号汇合，每publish一条数据，汇合中元素加1，每回调一次handleAck办法，unconfirm汇合删掉相应的一条（multiple=false）或多条（multiple=true）记录。从程序运行效率上看，这个unconfirm汇合最好采纳有序汇合SortedSet存储构造。实际上，SDK中的waitForConfirms()办法也是通过SortedSet保护音讯序号的。
要害代码：
SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
//别忘这行代码
channel.confirmSelect();
//增加监听器
channel.addConfirmListener(new ConfirmListener() {
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
if (multiple) {
confirmSet.headSet(deliveryTag + 1).clear();
} else {
confirmSet.remove(deliveryTag);
}
}
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.out.println("Nack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
if (multiple) {
confirmSet.headSet(deliveryTag + 1).clear();
} else {
confirmSet.remove(deliveryTag);
}
}
});
while (true) {
long nextSeqNo = channel.getNextPublishSeqNo();
channel.basicPublish(ConfirmConfig.exchangeName, ConfirmConfig.routingKey, MessageProperties.PERSISTENT_TEXT_PLAIN, ConfirmConfig.msg_10B.getBytes());
confirmSet.add(nextSeqNo);
}
3、音讯回调
//要点
//1.发送音讯是将第三个参数mandatory设置为true
channel.basicPublish(EXCHANGE,FOOD,true,null,"肯德基提价了".getBytes());
//2.增加音讯回调监听器
channel.addReturnListener(new ReturnListener() {
@Override
public void handleReturn(int i, String s, String s1, String s2, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {
System.out.println("音讯不可路由"+new String(bytes,"utf-8"));
}
});
//留神：开启回调不能敞开连贯和信道，
2.4、接管方确认
1、概述
接管方ack分为手动和主动，在接管音讯时设置
//第二个参数就是指定是否手动ack false时为手动
channel.basicConsume(QUEUE_ONE,false,consumer);
手动ack有三种
- 单个确认
- 单个回绝
- 批量回绝
2、代码实现
单个确认ack
//接管音讯
Consumer consumer = new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println(new String(body,"utf-8"));
//手动应答ack能够在该办法中进行；参数：1.音讯tag，2.是否批量ack
channel.basicAck(envelope.getDeliveryTag(),false);
//回绝音讯;参数：1.音讯tag；2.音讯是否从新入队，当只有一个消费者时，会引起反复生产
channel.basicReject(envelope.getDeliveryTag(),false);
//批量ack音讯;参数：1.音讯tag；2.是否批量ack音讯，3.是否重回队列
channel.basicNack(envelope.getDeliveryTag(),true,false);
}
};
//这里只须要条应答的语句，我这里常识都列出来
channel.basicConsume(QUEUE_ONE,false,consumer);
//留神下面第二个参数要为false能力手动ack
2.5、音讯TTL和队列TTL、死信队列、提早队列
这一块临时不应用原始RabbitMq Client API实现，前面再钻研，然而会应用上面的org.springframework.amqp来实现
三、应用org.springframework.amqp操作mq
3.1、前言：
Spring对RabbitMp进行了形象，将交换机，队列，音讯，绑定，连贯等形象出实体类，不便操作，还提供了RabbitAdmit 和RabbitTemplate 来不便交换机队列的治理以及音讯的发送接管等
3.2、根本实例
0、环境和依赖
<!-- 环境
* jdk 1.8
* idea
* springboot 2.2.6
-->
<!-- 依赖 -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1、实例
发送音讯通知消费者超时打折了快来购物
- 配置类
@Configuration
public class RabbitConfig {
private final static Logger log = LoggerFactory.getLogger(RabbitConfig.class);
private final static String EXCHANGE_NAME = "verification_code_exchange";
private final static String VERIFICATION_CODE_QUEUE = "verification_code_queue";
private final static String VERIFICATION_CODE_ROUTE_KEY = "verification_code_key";
//死信交换机和队列和key
private final static String DLX_EXCHANGE_NAME = "dlx-exchange";
private final static String DLX_KEY = "verification_code_key";
@Bean
public CachingConnectionFactory connectionFactory(){
CachingConnectionFactory conn = new CachingConnectionFactory();
conn.setUsername("leyou");
conn.setPassword("leyou");
conn.setVirtualHost("/leyou");
//音讯发送到mq发送确认音讯给生产者
conn.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
//音讯发送到mq，通过绑定的key找不到queue，则发送音讯给生产者
conn.setPublisherReturns(true);
return conn;
}
@Bean
public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
//设置音讯序列化
rabbitTemplate.setMessageConverter(converter());
//音讯的确认回调
// rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
// @Override
// public void confirm(CorrelationData correlationData, boolean b, String s) {
//
// }
// });
rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
//ack为确认音讯是否胜利发送到mq
if(ack){
//胜利发送
log.info("音讯发送胜利");
}
});
//改标记位设置位true时，当交换机依据本身类型和routeKey无奈找到对应的队列时，
// 则mq会将音讯返还给生产者
//当为false时则mq会将音讯间接删除
rabbitTemplate.setMandatory(true);
//音讯，返回码，返回内容，交换机，路由key
rabbitTemplate.setReturnCallback((Message message, int replyCode, String replyText, String exchange, String routingKey)->{
//音讯
log.info("message:{},replyCode:{},replyText:{},exchange:{},routingKey:{}",message,replyCode,replyText,exchange,routingKey);
});
return rabbitTemplate;
}
/**
* 注入rabbitadmin 用来申明交换机和队列，次要作用是代替原始的应用channl申明的做法，全副交给这个对象来实现
* @param connectionFactory
* @return
*/
@Bean
public RabbitAdmin rabbitAdmit(CachingConnectionFactory connectionFactory){
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
return rabbitAdmin;
}
/**
* 音讯序化对象
* 默认应用的是JDK的序列化，这里配置了后就能够将音讯序列化为json格局
*/
@Bean
public MessageConverter converter() {
return new Jackson2JsonMessageConverter();
}
/**
* 申明一个交换机
*/
@Bean
public DirectExchange verificationCodeExchange(RabbitAdmin rabbitAdmin){
DirectExchange exchange = new DirectExchange(EXCHANGE_NAME);
rabbitAdmin.declareExchange(exchange);
return exchange;
}
/**
* 申明一个队列
* @param rabbitAdmin
* @return
*/
@Bean
public Queue getQueue(RabbitAdmin rabbitAdmin){
Queue queue = new Queue(VERIFICATION_CODE_QUEUE, true,false,false,null);
rabbitAdmin.declareQueue(queue);
return queue;
}
/**
* 申明一个绑定
* @param rabbitAdmin
* @param verificationCodeExchange
* @return
*/
@Bean
public Binding bindingQueue(RabbitAdmin rabbitAdmin,DirectExchange verificationCodeExchange){
Binding with = BindingBuilder.bind(getQueue(rabbitAdmin)).to(verificationCodeExchange).with(VERIFICATION_CODE_ROUTE_KEY);
rabbitAdmin.declareBinding(with);
return with;
}
}
阐明：下面用到了生产者confirm和音讯回调机制
1、生产者confirm要害代码：//1、创立连贯时 conn.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED); //2、创立rabbitTemplate时 rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> { //ack为确认音讯是否胜利发送到mq if(ack){ //胜利发送 log.info("音讯发送胜利"); } });2、音讯回调机制要害代码：
//1、创立连贯时 conn.setPublisherReturns(true); //2、创立rabbitTemplate时 //改标记位设置位true时，当交换机依据本身类型和routeKey无奈找到对应的队列时， // 则mq会将音讯返还给生产者 //当为false时则mq会将音讯间接删除 rabbitTemplate.setMandatory(true); rabbitTemplate.setReturnCallback((Message message, int replyCode, String replyText, String exchange, String routingKey)->{ //音讯 log.info("message:{},replyCode:{},replyText:{},exchange:{},routingKey:{}",message,replyCode,replyText,exchange,routingKey); });
生产者：
@Component
public class RabbitSender {
//注入rabbitTemplate
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendMsg(){
//构建音讯
Message message = MessageBuilder.withBody(
JSONObject.toJSONString(MessageModel.builder().id(msgId).context("超市打折，快来抢购！").build()).getBytes()).build();
//音讯长久化
message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
//音讯的媒体类型
message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);
//音讯的自定义关联id
CorrelationData correlationData = new CorrelationData(String.valueOf(msgId));
rabbitTemplate.convertAndSend(exchange,routingKey,message,new MessagePostProcessor(){
//音讯后置处理器，能够在上面这个办法中对音讯进行相干属性的设置
@Override
public Message postProcessMessage(Message message) throws AmqpException {
//比方能够设置下面 这些属性等
//message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);//音讯长久化问题 //message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);//音讯的媒体类型
return message;
}
},correlationData);
}
}
消费者
@Component
public class RabbitReceive {
@RabbitListener(bindings = @QueueBinding(exchange = @Exchange(value = EXCHANGE_NAME, type = ExchangeTypes.DIRECT),
key = VERIFICATION_CODE_ROUTE_KEY,
value = @Queue(value = VERIFICATION_CODE_QUEUE, autoDelete = "false"),
ignoreDeclarationExceptions = "true"),
concurrency = "1", // 指定监听该队列的消费者个数
ackMode = "MANUAL"// 手动ack
)
public void receiveCode(Channel channel, Message msg, @Headers Map<String, Object> headers) throws IOException, InterruptedException {
String msgId = (String) headers.get("spring_listener_return_correlation");
long tag = msg.getMessageProperties().getDeliveryTag();
channel.basicAck(tag, false);
}
}
其中：发送方确认（生产者confirm）、音讯回调下面代码都蕴含了；消费者ack则和原始办法是一样的
上面介绍音讯TTL，队列TTL，死信队列，提早队列
- 音讯和队列的TTL
//音讯ttl
//在构建音讯时设置音讯的过期工夫
Message message = MessageBuilder.withBody(
JSONObject.toJSONString(MessageModel.builder().id(msgId).context("超市打折，快来抢购！").build()).getBytes()).build();
//音讯的过期工夫
message.getMessageProperties().setExpiration("5000");
//队列的ttl
//在创立队列时通过参数设置
Map<String, Object> args = new HashMap<>();
//指定死信交换机
args.put("x-dead-letter-exchange", DLX_EXCHANGE_NAME);
//指定死信队列的key
args.put("x-dead-letter-routing-key", DLX_KEY);
//设置队列中音讯的过期工夫 ms
args.put("x-message-ttl",10000);
//整个队列的过期工夫，过期后整个队列会被删除
//args.put("x-expires",10000);
Queue queue = new Queue(VERIFICATION_CODE_QUEUE, true,false,false,args);
下面还包含死信队列的属性设置，和死信队列key，对于死信队列的配置，还须要配置一个死信交换机和一个死信队列；当有音讯或队列的ttl过期，音讯超过队列最大长度，音讯被回绝且设置不从新回队列，则音讯会被转发到死信交换机，再转发到死信队列。
- 对于提早队列的实现办法有两种
- 应用死信队列，用一个设置了ttl的队列来寄存音讯，该队列不须要消费者监听，而后给该队列配置死信交换机和队列，消费者监听死信队列，这样就能达到工夫达到提早收到音讯的目标
- 应用rabbitmq插件的形式实现，这里先不写，放到下一篇笔记中
最初
感激你看到这里，看完有什么的不懂的能够在评论区问我，感觉文章对你有帮忙的话记得给我点个赞，每天都会分享java相干技术文章或行业资讯，欢送大家关注和转发文章！
发表回复