关键词:
六大核心部分
Java代码实现HelloWorld
新建Maven工程,引入依赖
<!-- 指定 jdk 编译版本 -->
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>8</source>
<target>8</target>
</configuration>
</plugin>
</plugins>
</build>
<dependencies>
<!--rabbitmq 依赖客户端 -->
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.8.0</version>
</dependency>
<!-- 操作文件流的一个依赖 -->
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
<version>2.6</version>
</dependency>
</dependencies>
生产者代码书写
package com.vleus.rabbitmq.one;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author vleus
* @date 2021年07月19日 22:00
*/
public class Producer
//设置队列名称
public static final String QUEUE_NAME = "hello";
//发消息
public static void main(String[] args) throws IOException, TimeoutException
//创建一个连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//设置工厂IP,连接rabbitmq的队列
connectionFactory.setHost("192.168.37.139");
//设置用户名和密码
connectionFactory.setUsername("admin");
connectionFactory.setPassword("123456");
//创建连接
Connection connection = connectionFactory.newConnection();
//获取信道
Channel channel = connection.createChannel();
/**
* 生成一个队列:
* 1、队列名称;
* 2、队列里面的消息是否持久化(存储在磁盘中),默认情况下消息存储在内存中;
* 3、该队列是否只供一个消费者进行消费,是否进行消息共享,true可以多个消费者消费,false只能一个消费者消费
* 4、是否自动删除,最后一个消费者端开连接以后,该队列是否自动删除,true自动删除,false不自动删除;
* 5、其他参数
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//发消息
String message = "Hello,World";
/**
* 发布消息
* 1、发送到哪个交换机;
* 2、路由的key值是哪个,本次是队列名称,
* 3、其他参数信息;
* 4、发送消息的消息体
*/
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
System.out.println("消息发送完毕。。。");
消费者代码书写
package com.vleus.rabbitmq.one;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
/**
* @author vleus
* @date 2021年07月19日 22:13
* 消费者,接收消息的
*/
public class Consumer
//队列名称
public static final String QUEUE_NAME = "hello";
//接收消息
public static void main(String[] args) throws IOException, TimeoutException
//创建连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//设置工厂IP,连接rabbitmq的队列
connectionFactory.setHost("192.168.37.139");
//设置用户名和密码
connectionFactory.setUsername("admin");
connectionFactory.setPassword("123456");
//创建新的连接
Connection connection = connectionFactory.newConnection();
//创建信道
Channel channel = connection.createChannel();
//声明 接收消息的回调
DeliverCallback deliverCallback = (consumerTag,message) ->
System.out.println(new String(message.getBody()));
;
CancelCallback cancelCallback = consumerTag ->
System.out.println("消息消费被中断");
;
/**
* 消费者消费消息
* 1、消费哪个队列;
* 2、消费成功之后是否需要自动应答,true自动应答,false手动应答;
* 3、消费者未成功消费的回调;
* 4、消费者取消消费的回调;
*
*/
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
WorkQueues
工作队列(又称任务队列)的主要思想是避免立即执行资源密集型任务,而不得不等待它完成。相反我们安排任务在之后执行。我们把任务封装为消息并将其发送到队列。在后台运行的工作进程将弹出任务并最终执行作业。当有多个工作线程时,这些工作线程将一起处理这些任务。
轮训发送消息
代码实现如下:
抽取连接工厂工具类
package com.vleus.rabbitmq.utils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/**
* @author vleus
* @date 2021年07月19日 22:37
* 此类为连接工厂创建信道的工具类
*/
public class RabbitMqUtils
public static Channel getChannel() throws Exception
//创建一个连接工厂
ConnectionFactory connectionFactory = new ConnectionFactory();
//设置工厂IP,连接rabbitmq的队列
connectionFactory.setHost("192.168.37.139");
//设置用户名和密码
connectionFactory.setUsername("admin");
connectionFactory.setPassword("123456");
//创建连接
Connection connection = connectionFactory.newConnection();
//获取信道
Channel channel = connection.createChannel();
return channel;
设置工作线程数据
package com.vleus.rabbitmq.two;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import com.vleus.rabbitmq.utils.RabbitMqUtils;
/**
* @author vleus
* @date 2021年07月19日 22:41
* 这是一个工作线程,相当于之前的消费者
*/
public class Worker01
//设置队列名称
public static final String QUEUE_NAME = "hello";
//接收消息
public static void main(String[] args) throws Exception
Channel channel = RabbitMqUtils.getChannel();
//声明 接收消息的回调
DeliverCallback deliverCallback = (consumerTag, message) ->
System.out.println("接收到的消息为:" + new String(message.getBody()));
;
CancelCallback cancelCallback = consumerTag ->
System.out.println(consumerTag + "消息者取消消费接口回调逻辑");
;
//消息的接收
/**
* 消费者消费消息
* 1、消费哪个队列;
* 2、消费成功之后是否需要自动应答,true自动应答,false手动应答;
* 3、消费者未成功消费的回调;
* 4、消费者取消消费的回调;
*/
System.out.println("C2等待接收消息......");
channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);
设置生产者数据
package com.vleus.rabbitmq.two;
import com.rabbitmq.client.Channel;
import com.vleus.rabbitmq.utils.RabbitMqUtils;
import java.util.Scanner;
/**
* @author vleus
* @date 2021年07月19日 22:52
* 生产者,可以发送大量的消息
*/
public class Task01
//设置队列名称
public static final String QUEUE_NAME = "hello";
public static void main(String[] args) throws Exception
Channel channel = RabbitMqUtils.getChannel();
//声明队列
/**
* 生成一个队列:
* 1、队列名称;
* 2、队列里面的消息是否持久化(存储在磁盘中),默认情况下消息存储在内存中;
* 3、该队列是否只供一个消费者进行消费,是否进行消息共享,true可以多个消费者消费,false只能一个消费者消费
* 4、是否自动删除,最后一个消费者端开连接以后,该队列是否自动删除,true自动删除,false不自动删除;
* 5、其他参数
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
//从控制台中接收信息
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext())
String message = scanner.next();
/**
* 发布消息
* 1、发送到哪个交换机;
* 2、路由的key值是哪个,本次是队列名称,
* 3、其他参数信息;
* 4、发送消息的消息体
*/
channel.basicPublish("",QUEUE_NAME,null,message.getBytes());
System.out.println("发送消息完成:" + message);
效果,两个工作线程:
可以看出来,两个工作线程是轮训消费生产者发来的数据;
消息应答
概念
消费者完成一个任务可能需要一段时间,如果其中一个消费者处理一个长的任务并仅只完成了部分突然它挂掉了,会发生什么情况。RabbitMQ 一旦向消费者传递了一条消息,便立即将该消息标记为删除。在这种情况下,突然有个消费者挂掉了,我们将丢失正在处理的消息。以及后续发送给该消费这的消息,因为它无法接收到。
为了保证消息在发送过程中不丢失,rabbitmq 引入消息应答机制,消息应答就是: 消费者在接收到消息并且处理该消息之后,告诉rabbitmq 它已经处理了,rabbitmq 可以把该消息删除了;
自动应答
消息发送后立即被认为已经传送成功,这种模式需要在高吞吐量和数据传输安全性方面做权衡,因为这种模式如果消息在接收到之前,消费者那边出现连接或者 channel 关闭,那么消息就丢失了,当然另一方面这种模式消费者那边可以传递过载的消息, 没有对传递的消息数量进行限制,当然这样有可能使得消费者这边由于接收太多还来不及处理的消息,导致这些消息的积压,最终使得内存耗尽,最终这些消费者线程被操作系统杀死, 所以这种模式仅适用在消费者可以高效并以某种速率能够处理这些消息的情况下使用。
手动应答
- Channel.basicAck:用于肯定确认,RabbitMQ已知道该消息并且已经成功处理消息,可以将其丢弃了;
- Channel.basicNack:用于否定确认;
- Channel.basicReject:用于否定确认,与Channel.basicNack相比少了一个参数,不处理该消息了直接拒绝,可以将其丢弃了;
Multiple的解释
手动应答的好处是可以批量应答并且减少网路拥堵;
multiple的true和false代表不同意思:
- true代表批量应答channel上未应答的消息:比如说channel上有传送tag的消息5、6、7、8,那么此时5-8的这些还未应答的消息都会被确认收到消息应答;
- false同上面相比:只会应答tag=8的消息,5,6,7这三个消息依然不会被确认收到消息应答;
消息自动重新入队
如果消费者由于某些原因失去连接(其通道已关闭,连接已关闭或 TCP 连接丢失),导致消息未发送 ACK 确认,RabbitMQ 将了解到消息未完全处理,并将对其重新排队。如果此时其他消费者可以处理,它将很快将其重新分发给另一个消费者。这样,即使某个消费者偶尔死亡,也可以确保不会丢失任何消息。
消息手动应答代码
默认消息采用的是自动应答,所以我们要想实现消息消费过程中不丢失,需要把自动应答改为手动应答,消费者在上面代码的基础上增加下面画红色部分代码。
消息生产者:
学相伴狂神说rabbitmq笔记(简单使用rabbitmq)(代码片段)
目录什么是rabbitMQ使用docker安装RabbitMQ,如果没有使用过docker的可以看这篇文章https://blog.csdn.net/qq_44716544/article/details/119870837授权账号和密码设置用户分配操作权限RabbitMQ支持的消息模型1.入门案例1.RabbitMQ入门案例-Simple简单模式2.什... 查看详情
rabbitmq知识点总结(代码片段)
RabbitMQ知识点总结消息队列MQ的相关概念什么是MQ为什么要用MQ?MQ的分类MQ的选择RabbitMQ简介四大概念RabbitMQ核心部分六大模式RabbitMQ工作原理简单模式工作模式轮询分发消息消息应答自动应答手动应答消息自动重新入队RabbitMQ... 查看详情
rabbitmq学习笔记2:消息队列核心组成部分--协议持久化分发策略高可用高可靠
一、消息队列协议1.什么是协议?计算机底层操作系统和应用程序通讯时共同遵守的一组约定,只有遵循共同的约定和规范,系统和底层操作系统之间才能相互交流。和一般的网络应用程序的不同它主要负责数据的接受和传递,... 查看详情
rabbitmq---消息队列---上半部分(代码片段)
RabbitMQ初识MQ什么是MQ为什么要用MQ1.流量消峰2.应用解耦3.异步处理MQ的分类1.ActiveMQ2.Kafka3.RocketMQ4.RabbitMQMQ的选择1.Kafka2.RocketMQ3.RabbitMQRabbitMQRabbitMQ的概念四大核心概念生产者交换机队列消费者RabbitMQ核心部分各个名词介绍RabbitMQ安装... 查看详情
rabbitmq必备核心知识
现在很多知名的互联网公司都有用到RabbitMQ,其性能,可扩展性让很多大公司青睐于使用它,不过想要完全使用好RabbitMQ需要掌握其核心的一些概念,这里就说说掌握RabbitMQ所需的必要知识 生产者与消费者 生产者:创... 查看详情
rabbitmq超详细学习笔记(章节清晰+通俗易懂)(代码片段)
...学习,最近花了一段时间系统学习了当下最为主流的RabbitMQ消息队列,学习过程中也随时记录,刚开始学习的时候懵懵懂懂,做的笔记都比较杂乱,系统学习完后我将笔记内容不断反复修改,对章节进行设... 查看详情
springbootmq系列rabbitmq核心知识点小结
【MQ系列】RabbitMq核心知识点小结以下内容,部分取材于官方教程,部分来源网络博主的分享,如有兴趣了解更多详细的知识点,可以在本文最后的文章列表中获取原地址RabbitMQ是一个基于AMQP协议实现的企业级消息系统,想要顺... 查看详情
消息队列之rabbitmq
...iveMQ(二)Kafka(三)RocketMQ(四)RabbitMQ四、RabbitMQ(一)RabbitMQ后台管理界面(二)RabbitMQ的四大核心概念1.生产者2.交换机3.队列4.消费者(三)RabbitMQ的核心部分1.RabbitMQ的六大模式2.... 查看详情
30分钟学玩转rabbitmq
最近在学习RabbitMQ,在网上找了不少资料发现都特高端。动辄集群部署,分布式架构什么的,对于一个初学者实在不够友好。心想求人不如求自己,为什么不自己整理一套资料呢?于是《30分钟学玩转RabbitMQ》诞生。一、准备工作... 查看详情
rabbitmq系统学习
各种mqactivemq,kafka使用zookeeper做管理rocketmq自己实现nameserverbroke管理AMQP核心概念高级消息队列协议publisherapplication->Server->Virtualhost->Exchange->MessageQueue->ConsumerapplicationServer:又称Broker,接收客户端的 查看详情
rabbitmq学习
RabbitMQ是基于AMQP协议的,通过使用通用协议就可以做到在不同语言之间传递。 AMQP协议核心概念server:又称broker,接受客户端连接,实现AMQP实体服务。connection:连接和具体broker网络连接。channel:网络信道,几乎所有操作都... 查看详情
rabbitmq学习笔记(自用)(代码片段)
...概念1.1什么是MQ1.2为什么要用MQ1.3MQ的分类1.4MQ的选择二、RabbitMQ2.1RabbitMQ的概念2.2四大核心概念三、简单案例3.1WorkQueues3.2轮训分发消息3.3消息应答3.4RabbitMQ持久化3.5不公平分发3.6预取值分发四、RabbitMQ-发布确认4.1发布确认逻辑4.2发... 查看详情
rabbitmq学习--交换机学习(第三个部分:发布-订阅模式)(代码片段)
RabbitMQ交换机学习在之前的学习中,我们是创建了一个工资队列。我们假设的是在工作队列背后,每个人物都恰好交付给一个确认的消费者(工作进程)。在本节的学习中,我们将做一些不同的事情:将消... 查看详情
rabbitmq学习--交换机学习(第三个部分:发布-订阅模式)(代码片段)
RabbitMQ交换机学习在之前的学习中,我们是创建了一个工资队列。我们假设的是在工作队列背后,每个人物都恰好交付给一个确认的消费者(工作进程)。在本节的学习中,我们将做一些不同的事情:将消... 查看详情
节清晰+通俗易懂)(代码片段)
...学习,最近花了一段时间系统学习了当下最为主流的RabbitMQ消息队列,学习过程中也随时记录,刚开始学习的时候懵懵懂懂,做的笔记都比较杂乱,系统学习完后我将笔记内容不断反复修改,对章节进行设... 查看详情
rabbitmq学习笔记2-理解消息通信
...的相关信息,包含具体的交换器、消息的接受兴趣方等。rabbitmq的基础流程如下:RabbitMQ的客户端和服务端通过channel与RabbitMQ服务器进行通信。Channel(信道):程序和RabbitMQ之间的连接是通过channel,channel是基于TCP协议之上的?,一个... 查看详情
rabbitmq核心功能介绍(代码片段)
RabbitMQ核心功能一.MQ的概念与功能介绍二.RabbitMQ的介绍和入门案例三.RabbitMQ的工作队列四.RabbitMQ的工作模式五.RabbitMQ的发布确认六.RabbitMQ的死性队列七.RabbitMQ的延迟队列本文对RabbitMQ核心功能的介绍,没有介绍RabbitMQ的安装与... 查看详情
rabbitmq核心功能介绍(代码片段)
RabbitMQ核心功能一.MQ的概念与功能介绍二.RabbitMQ的介绍和入门案例三.RabbitMQ的工作队列四.RabbitMQ的工作模式五.RabbitMQ的发布确认六.RabbitMQ的死性队列七.RabbitMQ的延迟队列本文对RabbitMQ核心功能的介绍,没有介绍RabbitMQ的安装与... 查看详情