组成部分说明:
RabbitMQ包括五种队列模式,简单队列、工作队列、发布/订阅、路由、主题。
生产者将消息发送到队列,消费者从队列获取消息。
连接工具类:
/**
* 连接工厂创建的工具类
*/
public class RabbitMqUtils {
public static Channel getChannel() throws Exception{
//创建一个工厂
ConnectionFactory factory = new ConnectionFactory();
//工厂ip 连接RabbitMQ的队列
factory.setHost("192.168.75.128");
//用户名
factory.setUsername("guest");
//密码
factory.setPassword("guest");
//创建连接
Connection connection = factory.newConnection();
//获取通道
return connection.createChannel();
}
}
生产者发送消息:
public class Send {
private final static String QUEUE_NAME = "simple_queue";
public static void main(String[] argv) throws Exception {
// 1、获取到通道
Channel channel = RabbitMqUtils.getChannel();
// 2、声明(创建)队列
/**
* 参数明细
* 1、queue 队列名称
* 2、durable 是否持久化,如果持久化,mq重启后队列还在
* 3、exclusive 是否独占连接,队列只允许在该连接中访问,
* 如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
* 4、autoDelete 自动删除,队列不再使用时是否自动删除此队列,
* 如果将此参数和exclusive参数设置为true就可以实现临时队列(队列不用了就自动删除)
* 5、arguments 参数,可以设置一个队列的扩展参数,比如:可设置存活时间
*/
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 3、消息内容
String message = "Hello World!";
// 4、向指定的队列中发送消息
/**
* 参数明细:
* 1、exchange,交换机,如果不指定将使用mq的默认交换机(设置为"")
* 2、routingKey,路由key,交换机根据路由key来将消息转发到指定的队列,
* 如果使用默认交换机,routingKey设置为队列的名称
* 3、props,消息的属性
* 4、body,消息内容
*/
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" 简单模式发送信息 '" + message + "'");
}
}
控制台结果:
web管理页面:
消费者接收消息:
public class Receive {
private final static String QUEUE_NAME = "simple_queue";
public static void main(String[] argv) throws Exception {
//获取通道
Channel channel = RabbitMqUtils.getChannel();
//接收到消息回调此函数
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
* 当接收到消息后此方法将被调用
* @param consumerTag 消费者标签,用来标识消费者的,在监听队列时设置channel.basicConsume
* @param envelope 信封,通过envelope
* @param properties 消息属性
* @param body 消息内容
* @throws IOException
*/
@Override
public 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();
// body 即消息体
String msg = new String(body, StandardCharsets.UTF_8);
System.out.println(" 简单模式收到信息 : "+msg+", 交换机:"+exchange+", 消息id: "+deliveryTag);
}
};
// 监听队列,第二个参数:是否自动进行消息确认。
/**
* 参数明细:
* 1、queue 队列名称
* 2、autoAck 自动回复,当消费者接收到消息后要告诉mq消息已接收,
* 如果将此参数设置为tru表示会自动回复mq,如果设置为false需要手动应答
* 3、callback,消费方法,当消费者接收到消息要执行的方法
*/
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
控制台结果:
web管理页面:
两个消费端共同消费同一个队列中的消息,但是一个消息只能被一个消费者获取。
生产者生产多个消息:
/**
* 生产者 生产大量消息
*/
public class Task01 {
private static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMqUtils.getChannel();
channel.queueDeclare(QUEUE_NAME,false,false,false,null);
//从控制台当中接受信息
Scanner scanner = new Scanner(System.in);
while (scanner.hasNext()){
String message = scanner.next();
channel.basicPublish("",QUEUE_NAME,null, message.getBytes());
System.out.println("发送消息完成:"+message);
}
}
}
消费者启动两个线程:
/**
* 工作线程
*/
public class Worker01 {
//队列名称
public static final String QUEUE_NAME = "work_queue";
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+"消费者取消消费接口回调逻辑");
};
System.out.println("消费者1/2 等待接受消息...");
channel.basicConsume(QUEUE_NAME,true,deliverCallback,cancelCallback);
}
}
控制台:
根据结果发现,默认情况,mq是轮询分发的
通过 BasicQos 方法设置prefetchCount = 1。这样RabbitMQ就会使得每个Consumer在同一个时间点最多处理1个Message。换句话说,在接收到该Consumer的ack前,他它不会将新的Message分发给它。相反,它会将其分派给不是仍然忙碌的下一个Consumer。
prefetchCount在手动ack的情况下才生效,自动ack不生效。
x:交换机类型:Fanout,也称为广播
发送消息到交换机,交换机默认绑定到每个队列中,故所有消息都会发送到每个通道,实现一个消息被多个消费者消费。
生产者:
/**
* 声明Exchange,不再声明Queue
* 消费者中声明Queue
*/
public class Send {
private final static String EXCHANGE_NAME = "fanout_exchange";
public static void main(String[] argv) throws Exception {
// 获取通道
Channel channel = RabbitMqUtils.getChannel();
// 声明exchange,指定类型为fanout
channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
// 消息内容
String message = "注册成功";
// 发布消息到Exchange
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
System.out.println("生产者广播信息 '" + message + "'");
}
}
消费者1:
public class Receive {
private final static String QUEUE_NAME = "fanout_exchange_queue1";//消费者1队列
private final static String EXCHANGE_NAME = "fanout_exchange";
public static void main(String[] argv) throws Exception {
// 获取通道
Channel channel = RabbitMqUtils.getChannel();
// 声明队列
channel.queueDeclare(QUEUE_NAME, false, false, false, null);
// 绑定队列到交换机
channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
// 定义队列的消费者
DefaultConsumer consumer = new DefaultConsumer(channel) {
// 获取消息,并且处理,这个方法类似事件监听,如果有消息的时候,会被自动调用
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
byte[] body) throws IOException {
System.out.println(" 消费者1收到信息 : " + new String(body) + "!");
}
};
// 监听队列,自动返回完成
channel.basicConsume(QUEUE_NAME, true, consumer);
}
}
消费者2和1类似。
控制台:
生产者发送消息,消费者1、2都可以收到消息。
与路由模式不同的是,路由模式发送者只能精准匹配一个routing key,而主题模式可以实现发送者匹配多个队列
文章浏览阅读9.6k次,点赞40次,收藏215次。一、环境介绍MCU: STM32F103ZET6代码开发工具: Keil5TCP/IP协议栈: LWIP网卡: DM9000本篇文章主要讲解如何在STM32F103工程里添加移植LWIP协议,最终完成TCP服务器、TCP客户端的通信测试。 网卡采用的是DM9000,工程代码中,采用STM32的FSMC接口来驱动DM900网卡,DM9000是并口网卡,引脚多,但是速度快,也可以采用其他网卡,SPI协议的、UART协议的等。 比如:W5500。 因为主要是讲LWIP协议栈的移植,所以网卡相关_网卡芯片和stm32
文章浏览阅读975次,点赞5次,收藏11次。仓库地址:仓库拉取:不想拉取可以直接下载压缩包。_opencv github
文章浏览阅读2.3k次。8种免费商用中文字体 转自https://www.uisdc.com/200-models-free-commercial-fonts 提取连接:https://pan.baidu.com/s/1nLxPRDDS1BTzBtISUfzaAA 提取码: tnxd 一、思源字体系列 大名鼎鼎的思源字体由 Adobe 在线字体库 Typekityu 与谷歌一起合作开发,字库对中(简体/繁体)、日、韩三国汉字进行了全面优化支持,可以说是非常全面了,其实思源不仅是个免费字体,还是一个开源字体,任何用户都可以_类似阿里汉仪智能黑体
文章浏览阅读1.9k次。有时候新导入一个项目,发现有这样一个问题 Error:Failed to open zip file. Gradle’s dependency cache may be > corrupt (this sometimes occurs after a network connection timeout.) href=”syncProject”>Re-download dependencies_androidstudio error: failed to open zip file. gradle's dependency cache may
文章浏览阅读1.6k次。李宏毅2021/2022春机器学习课程【《2021机器学习-李宏毅》学习笔记】一、机器学习1.机器学习基本概念找一个函数。深度学习:使用类神经网络的函数。可以有各式各样的输入:向量,矩阵(如图像),序列(如语音,文本);也可以有各式各样的输出:数值(regression),类别(classification),文本图像…函数类别Regression(回归):输出是一个数值(scalar)Classification(分类):给定选项/类别(classes),输出正确的Structur_李宏毅机器学习笔记
文章浏览阅读5.8w次,点赞2次,收藏3次。最近,要把原来基于Ubuntu下配置的Hadoop集群迁移到RHEL 5下,结果在启动的时候,出现了莫名其妙的问题:Namenode进程启动起来了,但是在登录到Datanode上启动集群slaves的时候,出现了Datanode无法连接到Namenode的问题,根据Datanode日志,可以很容易地定位到问题,肯定是在集群启动之初,Datanode向Namenode注册的时候失败了,在Data_rhel5.3 hadoop
文章浏览阅读3.1k次,点赞6次,收藏74次。高级计量经济学及stata应用 陈强第十五章 短面板15.13短面板的stata命令及实例第十五章 短面板15.13短面板的stata命令及实例clear all cd "F:\stata经济数据分析"diruse traffic.dta, clear* 1 面板数据的设定* xtset panelvar timevar* encode country, gen(cntry) //假如panelvar是字符串,转换为数字型xtset state year* 显示数据结构* x_stata研究啤酒税与交通死亡率
文章浏览阅读806次,点赞15次,收藏15次。我个人认为,如果你想靠着背面试题来获得心仪的offer,用癞蛤蟆想吃天鹅肉形容完全不过分。想必大家能感受到面试越来越难,想找到心仪的工作也是越来越难,高薪工作羡慕不来,却又对自己目前的薪资不太满意,工作几年甚至连一个应届生的薪资都比不上,终究是错付了,错付了自己没有去提升技术。这些面试题分享给大家的目的,其实是希望大家通过大厂面试题分析自己的技术栈,给自己梳理一个更加明确的学习方向,当你准备好去面试大厂,你心里有底,大概知道面试官会问多广,多深,避免面试的时候一问三不知。
文章浏览阅读988次。【代码】[学习笔记]python-anaconda安装PCV库。_python pcv
文章浏览阅读770次,点赞16次,收藏10次。是Android系统中一个非常重要的类,它代表了屏幕上发生的触摸事件。当用户在屏幕上触摸、滑动或者长按时,都会生成一个对象,这个对象包含了触摸动作的各种信息。
文章浏览阅读341次,点赞3次,收藏8次。历时半年,我们终于整理出了这份市面上最全面的最新Android面试题解析大全!
文章浏览阅读41次。中文乱码异常处理问题描述:数据库健康巡检时,执行脚本导出来的数据中文乱码处理方案:[oracle@rac1~]$ export NLS_LANG="SIMPLIFIED CHINESE"_CHINA.ZHS16GBK[oracle@rac1~]$ export LANG=zh_CN.GBK此后脚本执行出来的结果中文正常显示.注意export命令设置环境变量,只对当前bash登陆的session有效,它是存在内存里面的.可以编辑bash_profile文件进行永久设置vi.ba_oracle中文乱码数据过滤