1_RabbitMQ初入门入门Hello消费者+生产者
文章目录
- 1_RabbitMQ初入门
- 1.RabbitMQ的介绍
- 1.工作原理&发送/接收消息的流程
- 2. 为什么要使用消息队列?
- 3.使用了消息队列有什么缺点?
- 2.安装RabbitMQ
- 3.入门程序Hello_消费者&生产者_
- 1.导入依赖
- 2.生产者
- 1.设置连接信息
- 2.获取connection(连接),channel(信道)
- 3.channel(信道)绑定队列&实现消费方法
- 4.接收消息
- 完整代码:
- 3.消费者
- 1.设置连接信息
- 2.获取connection(连接),channel(信道)
- 3.channel(信道)绑定队列
- 4.发送消息
- 完整代码:
1_RabbitMQ初入门
1.RabbitMQ的介绍
1.工作原理&发送/接收消息的流程
Broker:消息队列服务进程,此进程包括两个部分:Exchange和Queue。
Exchange:消息队列交换机,按一定的规则将消息路由转发到某个队列,对消息进行过虑。
Queue:消息队列,存储消息的队列,消息到达队列并转发给指定的消费方。
Producer:消息生产者,即生产方客户端,生产方客户端将消息发送到MQ。
Consumer:消息消费者,即消费方客户端,接收MQ转发的消息。
消息发布接收流程:
-----发送消息----- 1、生产者和Broker建立TCP连接。
2、生产者和Broker建立通道。
3、生产者通过通道消息发送给Broker,由Exchange将消息进行转发。
4、Exchange将消息转发到指定的Queue(队列)----接收消息----- 1、消费者和Broker建立TCP连接
2、消费者和Broker建立通道
3、消费者监听指定的Queue(队列)
4、当有消息到达Queue时Broker默认将消息推送给消费者。
5、消费者接收到消息。
2. 为什么要使用消息队列?
解耦,异步,削峰
3.使用了消息队列有什么缺点?
分析:一个使用了MQ的项目,如果连这个问题都没有考虑过,就把MQ引进去了,那就给自己的项目带来了风险。我们引入一个技术,要对这个技术的弊端有充分的认识,才能做好防御。
系统的可用性降低
:如果消息队列挂了,那么系统也会受到影响
系统的复杂性增加
:要多考虑很多方面的问题,比如一致性问题、如何保证消息不被重复消费,如何保证消息可靠传输。因此,需要考虑的东更多,系统的复杂性增大。
2.安装RabbitMQ
详情请参考此链接
3.入门程序Hello_消费者&生产者_
1.导入依赖
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId>
</dependency><!-- 这个是日志包的依赖 -->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-logging</artifactId>
</dependency>
2.生产者
1.设置连接信息
通过ConnectionFactory获取连接信息
//通过连接工厂创建新的连接和mq建立连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);//端口
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
//设置虚拟机,一个mq服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mq
connectionFactory.setVirtualHost("/");
2.获取connection(连接),channel(信道)
Connection connection = null;
Channel channel = null;
//建立新连接
connection = connectionFactory.newConnection();
//创建会话通道,生产者和mq服务所有通信都在channel通道中完成
channel = connection.createChannel();
3.channel(信道)绑定队列&实现消费方法
//队列
private static final String QUEUE = "helloworld";//监听队列
//声明队列,如果队列在mq 中没有则要创建
//参数:String queue, boolean durable, boolean exclusive, boolean autoDelete, M
/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化,如果持久化,mq重启后队列还在* 3、exclusive 是否独占连接,队列只允许在该连接中访问,如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时* 4、autoDelete 自动删除,队列不再使用时是否自动删除此队列,如果将此参数和exclusive参数设置为true就可以实现临时队列(队列不* 5、arguments 参数,可以设置一个队列的扩展参数,比如:可设置存活时间*/
channel.queueDeclare(QUEUE,true,false,false,null);//实现消费方法
DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 当接收到消息后此方法将被调用* @param consumerTag 消费者标签,用来标识消费者的,在监听队列时设置channel.basicConsume* @param envelope 信封,通过envelope* @param properties 消息属性* @param body 消息内容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.B//交换机String exchange = envelope.getExchange();//消息id,mq在channel中用来标识消息的id,可用于确认消息已接收long deliveryTag = envelope.getDeliveryTag();//消息内容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}
};
4.接收消息
//监听队列//参数:String queue, boolean autoAck, Consumer callback/*** 参数明细:* 1、queue 队列名称* 2、autoAck 自动回复,当消费者接收到消息后要告诉mq消息已接收,如果将此参数设置为tru表示会自动回复mq,如果设置为false要通过编程实现回复* 3、callback,消费方法,当消费者接收到消息要执行的方法*/channel.basicConsume(QUEUE,true,defaultConsumer);
结果截图:
完整代码:
package com.xuecheng.test.rabbitmq;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer01 {//队列private static final String QUEUE = "helloworld";public static void main(String[] args) throws IOException, TimeoutException {//通过连接工厂创建新的连接和mq建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟机,一个mq服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mqconnectionFactory.setVirtualHost("/");//建立新连接Connection connection = connectionFactory.newConnection();//创建会话通道,生产者和mq服务所有通信都在channel通道中完成Channel channel = connection.createChannel();//监听队列//声明队列,如果队列在mq 中没有则要创建//参数:String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化,如果持久化,mq重启后队列还在* 3、exclusive 是否独占连接,队列只允许在该连接中访问,如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建* 4、autoDelete 自动删除,队列不再使用时是否自动删除此队列,如果将此参数和exclusive参数设置为true就可以实现临时队列(队列不用了就自动删除)* 5、arguments 参数,可以设置一个队列的扩展参数,比如:可设置存活时间*/channel.queueDeclare(QUEUE,true,false,false,null);//实现消费方法DefaultConsumer defaultConsumer = new DefaultConsumer(channel){/*** 当接收到消息后此方法将被调用* @param consumerTag 消费者标签,用来标识消费者的,在监听队列时设置channel.basicConsume* @param envelope 信封,通过envelope* @param properties 消息属性* @param body 消息内容* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {//交换机String exchange = envelope.getExchange();//消息id,mq在channel中用来标识消息的id,可用于确认消息已接收long deliveryTag = envelope.getDeliveryTag();//消息内容String message= new String(body,"utf-8");System.out.println("receive message:"+message);}};//监听队列//参数:String queue, boolean autoAck, Consumer callback/*** 参数明细:* 1、queue 队列名称* 2、autoAck 自动回复,当消费者接收到消息后要告诉mq消息已接收,如果将此参数设置为tru表示会自动回复mq,如果设置为false要通过编程实现回复* 3、callback,消费方法,当消费者接收到消息要执行的方法*/channel.basicConsume(QUEUE,true,defaultConsumer);}
}
3.消费者
1.设置连接信息
通过ConnectionFactory获取连接信息
//通过连接工厂创建新的连接和mq建立连接
ConnectionFactory connectionFactory = new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);//端口
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
//设置虚拟机,一个mq服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mq
connectionFactory.setVirtualHost("/");
2.获取connection(连接),channel(信道)
Connection connection = null;
Channel channel = null;
//建立新连接
connection = connectionFactory.newConnection();
//创建会话通道,生产者和mq服务所有通信都在channel通道中完成
channel = connection.createChannel();
3.channel(信道)绑定队列
//队列
private static final String QUEUE = "helloworld";//声明队列,如果队列在mq 中没有则要创建
//参数:String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化,如果持久化,mq重启后队列还在* 3、exclusive 是否独占连接,队列只允许在该连接中访问,如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建* 4、autoDelete 自动删除,队列不再使用时是否自动删除此队列,如果将此参数和exclusive参数设置为true就可以实现临时队列(队列不用了就自动删除)* 5、arguments 参数,可以设置一个队列的扩展参数,比如:可设置存活时间*/
channel.queueDeclare(QUEUE,true,false,false,null);
4.发送消息
//参数:String exchange, String routingKey, BasicProperties props, byte[] bod
/*** 参数明细:* 1、exchange,交换机,如果不指定将使用mq的默认交换机(设置为"")* 2、routingKey,路由key,交换机根据路由key来将消息转发到指定的队列,如果使用默认交换机,routingKey设置为队列的名称* 3、props,消息的属性* 4、body,消息内容*/
//消息内容
String message = "hello world 黑马程序员";
channel.basicPublish("",QUEUE,null,message.getBytes());
System.out.println("send to mq "+message);
结果截图:
完整代码:
package com.xuecheng.test.rabbitmq;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 Producer01 {//队列private static final String QUEUE = "helloworld";public static void main(String[] args) {//通过连接工厂创建新的连接和mq建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("127.0.0.1");connectionFactory.setPort(5672);//端口connectionFactory.setUsername("guest");connectionFactory.setPassword("guest");//设置虚拟机,一个mq服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mqconnectionFactory.setVirtualHost("/");Connection connection = null;Channel channel = null;try {//建立新连接connection = connectionFactory.newConnection();//创建会话通道,生产者和mq服务所有通信都在channel通道中完成channel = connection.createChannel();//声明队列,如果队列在mq 中没有则要创建//参数:String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*** 参数明细* 1、queue 队列名称* 2、durable 是否持久化,如果持久化,mq重启后队列还在* 3、exclusive 是否独占连接,队列只允许在该连接中访问,如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建* 4、autoDelete 自动删除,队列不再使用时是否自动删除此队列,如果将此参数和exclusive参数设置为true就可以实现临时队列(队列不用了就自动删除)* 5、arguments 参数,可以设置一个队列的扩展参数,比如:可设置存活时间*/channel.queueDeclare(QUEUE,true,false,false,null);//发送消息//参数:String exchange, String routingKey, BasicProperties props, byte[] body/*** 参数明细:* 1、exchange,交换机,如果不指定将使用mq的默认交换机(设置为"")* 2、routingKey,路由key,交换机根据路由key来将消息转发到指定的队列,如果使用默认交换机,routingKey设置为队列的名称* 3、props,消息的属性* 4、body,消息内容*///消息内容String message = "hello world 黑马程序员";channel.basicPublish("",QUEUE,null,message.getBytes());System.out.println("send to mq "+message);} catch (Exception e) {e.printStackTrace();} finally {//关闭连接//先关闭通道try {channel.close();} catch (IOException e) {e.printStackTrace();} catch (TimeoutException e) {e.printStackTrace();}try {connection.close();} catch (IOException e) {e.printStackTrace();}}}
}
1_RabbitMQ初入门入门Hello消费者+生产者相关推荐
- Spring Cloud入门-Ribbon服务消费者(Hoxton版本)
文章目录 Spring Cloud入门系列汇总 摘要 Ribbon简介 RestTemplate的使用 GET请求方法 getForObject方法 getForEntity方法 POST请求方法 p ...
- Hadoop快速入门——入门考试(伪分布式60+编码25+执行jar15)
Hadoop快速入门--入门考试(伪分布式60+编码25+执行jar15) 目录 一.伪分布式搭建(60分) 1.创建1台Linux虚拟机,并打开对应的网络连接(VMnet8)(5分) 2.通过xsh ...
- 【SEUSE】操作系统实验:消费者-生产者问题
操作系统实验:消费者-生产者问题 README 一.实验目的 二.实验内容 三.实验步骤 四.主要数据结构及其说明 五.程序运行时的初值和运行结果 六.实验体会 七.源程序 README 本实验报告仅 ...
- SpringCloud+CloudAlibaba微服务初阶入门
内容目录 微服务 为什么要用微服务? SpringCloud生态(旧版) 项目搭建 规范的创建父工程 父模块的pom的常用依赖 创建子模块 1.建module(maven) 2.改pom 3.写Yml ...
- Spring Cloud入门 -- Ribbon服务消费者(Hoxton.SR5版)
Ribbon简介 Ribbon是Netflix公司开源的客户端负载均衡器,可以控制Http和Tcp客户端的负载均衡. Ribbon默认提供了很多负载均衡算法,如:轮询.随机等,也可以实现自定义的负载均 ...
- Kafka快速入门(Kafka消费者)
Kafka 消费者 1. Kafka 消费方式 2 Kafka 消费者工作流程 2.1 消费者总体工作流程 2.2 消费者组原理 Consumer Group(CG):消费者组,由多个consumer ...
- C++初阶——入门(3)
目录 7. 内联函数 7.1 概念 7.2 特性 宏的优缺点? 8. auto关键字(C++11) 8.1 类型别名思考 8.2 auto简介 8.3 auto的使用细则 8.3 auto不能推导的场 ...
- 多线程锁,线程池,消费者生产者模型
锁是怎么存在的? 在python建立之初,开发者目的是为了快速把语言开发出来,如果加上GIL(c语言加锁),切换时按照100条字节指令来进行线程间的切换 为什么加锁? 非线程安全,控制一段代码 1.l ...
- kafka消费者如何读同一生产者消息_Kafka消费者生产者实例
为了更为直观展示Kafka的消息生产消费的过程,我会从基于Console和基于Application两个方面介绍使用实例.Kafka是一个分布式流处理平台,具体来说有三层含义: 它允许发布和订阅记录流 ...
最新文章
- zabbix详细介绍及其自动动态发现
- CRUX下实现进程隐藏(3)
- linux eof tmp目录,Linux学习-标准输入输出(示例代码)
- GitHub发布重大更新,关系到所有程序员!
- 方舟原始恐惧mod生物代码_主播在《方舟生存进化》成功打造幽灵城堡,可怕的怪物在这里横行...
- bzoj千题计划241:bzoj3864: Hero meet devil
- the python interpreter is in_the python interpreter is in - 百度学术
- 配置Ubuntu系统环境变量
- 【CAS】Implementing generic double-word compare and swap for x86/x86-64
- 4.4 Spark SQL实现用户ip地址热度分析
- VMware虚拟机下Ubuntu安装VMware Tools详解
- 机器学习实战1-1 KNN电影分类遇到的问题
- 穿越沙漠问题c语言算法,沙漠穿越问题_c语言课程设计.doc
- Base64 - 维基百科,自由的百科全书
- 《人生七年》纪录片中问的问题
- Ubuntu 中文字体美化方案大全 (3): 使用Windows XP字体
- java反射学习---(框架设计的灵魂,你不收藏都后悔)
- 挂耳耳机哪个牌子好,盘点五款挂耳式蓝牙耳机推荐
- Photoshop png图片改变 透明背景之外的图片颜色
- springboot 自定义日期出入参
热门文章
- [Hands On ML] 7. 集成学习和随机森林
- 端口可以随便设置吗_驱动可以随便更新吗?
- php.ini开启命名空间,Zend Framework教程之模型Model基本规则和使用方法
- 程序闪退怎么运行_苹果应用程序崩溃闪退怎么办?如何解决苹果设备的软故障?...
- python中的深浅拷贝
- try…finally与嵌套及自定义异常抛出
- 计算所有1-100之间数字之和(偶数之和)代码
- LeetCode第 227 场周赛题解
- Linux 程 序 员 失 业 警 告
- 最全BAT前端开发面试80题:算法+html+js+css!含答案大赠送!