Rabbitmq的五种模式和案例_rabbit设计模式-程序员宅基地
消费者监听队列,如果队列中有消息,就消费掉,消息被拿走后,自动从队列删除
(隐患,消息可能没有被消费者正确处理,已经消失了,无法恢复)
应用场景:聊天室
案例：
1>.首先准备依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>2>.写一个test类
public class SimpleTest {
//模拟生产者将消息放入队列
@Test
public void send() throws Exception{
/*1 创建连接工厂
* 2 配置共创config
* 3 获取连接
* 4获取信道
* 5 从信道声明queue
* 6 发送消息
* 7 释放资源
*/
ConnectionFactory factory=new ConnectionFactory();
factory.setHost("106.23.34.56");
factory.setPort(5672);
factory.setVirtualHost("/tb");
factory.setUsername("admin");
factory.setPassword("123456");
//从工厂获取连接
Connection conn=factory.newConnection();
//从连接获取信道
Channel chan=conn.createChannel();
//利用channel声明第一个队列
chan.queueDeclare("simple", false, false, false, null);
//queue String类型,表示声明的queue对列的名字
//durable Boolean类型,表示是否持久化
//exclusive Boolean类型:当前声明的queue是否专注;true当前连接创建的
//任何channle都可以连接这个queue,false,新的channel不可使用
//autoDelete Boolean类型:在最后连接使用完成后,是否删除队列,false
//arguments Map类型,其他声明参数
//发送消息
String msg="helloworld,nihaoa";
chan.basicPublish("", "simple", null, msg.getBytes());
//exchange String类型,交换机名称,简单模式使用默认交换""
//routingkey String类型,当前的消息绑定的routingkey,简单模式下,与队列同名即可
//props BasicProperties类型,消息的属性字段对象,例如BasicProperties
//可以设置一个deliveryMode的值0 持久化,1 表示不持久化,durable配合使用
//body byte[] :消息字符串的byte数组
}
//模拟消费端
@Test
public void receive() throws Exception{
ConnectionFactory factory=new ConnectionFactory();
factory.setHost("106.23.34.56");
factory.setPort(5672);
factory.setVirtualHost("/tb");
factory.setUsername("admin");
factory.setPassword("123456");
//从工厂获取连接Connection conn=factory.newConnection();//从连接获取信道Channel chan=conn.createChannel();chan.queueDeclare("simple", false, false, false, null);//创建一个消费者QueueingConsumer consumer= new QueueingConsumer(chan);chan.basicConsume("simple", consumer);//监听队列while(true){//获取下一个delivery,delivery从队列获取消息Delivery delivery = consumer.nextDelivery();String msg=new String(delivery.getBody());System.out.println(msg);}}}
2.work模式
生产者将消息放入队列
多个消费者同时监听同一个队列,消息如何被消费?
C1,C2共同争抢当前消息队列的内容,谁先拿到消息,谁来负责消费
应用场景:红包;大型项目中的资源调度过程(直接由最空闲的系统争抢到资源处理任务)
案例：
1>首先写一个工具类
public class ConnectionUtil {
public static Connection getConn(){
try{
ConnectionFactory factory=new ConnectionFactory();
factory.setHost("106.33.44.179");
factory.setPort(5672);
factory.setVirtualHost("/tb");
factory.setUsername("admin");
factory.setPassword("123456");
//从工厂获取连接
Connection conn=factory.newConnection();
return conn;
}catch(Exception e){
System.out.println(e.getMessage());
return null;
}
}
}2>写test类
public class WorkTest {
@Test
public void send() throws Exception{
//获取连接
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
//声明队列
chan.queueDeclare("work", false, false, false, null);
for(int i=0;i<100;i++){
String msg="1712,hello:"+i+"message";
chan.basicPublish("", "work", null, msg.getBytes());
System.out.println("第"+i+"条信息已经发送");
}
chan.close();
conn.close();
}
@Test
public void receive1() throws Exception{
//获取连接,获取信道
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
chan.queueDeclare("work", false, false, false, null);
//同一时刻服务器只发送一条消息给同一消费者,消费者空闲,才发送一条
chan.basicQos(1);
//定义消费者
QueueingConsumer consumer=new QueueingConsumer(chan);
//绑定队列和消费者的关系
//queue
//autoAck:消息被消费后,是否自动确认回执,如果false,不自动需要手动在
//完成消息消费后进行回执确认,channel.ack,channel.nack
//callback
//chan.basicConsume(queue, autoAck, callback)
chan.basicConsume("work", false, consumer);
//监听
while(true){
Delivery delivery=consumer.nextDelivery();
byte[] result = delivery.getBody();
String msg=new String(result);
System.out.println("接受到:"+msg);
Thread.sleep(50);
//返回服务器,回执
chan.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
@Test
public void receive2() throws Exception{
//获取连接,获取信道
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
chan.queueDeclare("work", false, false, false, null);
//同一时刻服务器只发送一条消息给同一消费者,消费者空闲,才发送一条
chan.basicQos(1);
//定义消费者
QueueingConsumer consumer=new QueueingConsumer(chan);
//绑定队列和消费者的关系
//queue
//autoAck:消息被消费后,是否自动确认回执,如果false,不自动需要手动在
//完成消息消费后进行回执确认,channel.ack,channel.nack
//callback
//chan.basicConsume(queue, autoAck, callback)
chan.basicConsume("work", false, consumer);
//监听
while(true){
Delivery delivery=consumer.nextDelivery();
byte[] result = delivery.getBody();
String msg=new String(result);
System.out.println("接受到:"+msg);
Thread.sleep(150);
//返回服务器,回执
chan.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
}3 publish/fanout发布订阅
生产者将消息交给交换机
有交换机根据发布订阅的模式设定将消息同步到所有的绑定队列中;
后端的消费者都能拿到消息
应用场景:邮件群发,群聊天,广告
案例：
public class FanoutTest {
//交换机,有类型,发布订阅:fanout
//路由模式:direct
//主题模式:topic
@Test
public void send() throws Exception {
//获取连接
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
//声明交换机
//参数意义,1 交换机名称,2 类型:fanout,direct,topic
chan.exchangeDeclare("fanoutEx", "fanout");
//发送消息
for(int i=0;i<100;i++){
String msg="1712 hello:"+i+"msg";
chan.basicPublish("fanoutEx", "", null, msg.getBytes());
System.out.println("第"+i+"条信息已经发送");
}
}
@Test
public void receiv01() throws Exception{
//获取连接
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
//生命队列
chan.queueDeclare("fanout01", false, false, false, null);
//声明交换机
chan.exchangeDeclare("fanoutEx", "fanout");
//绑定队列到交换机
//参数 1 队列名称,2 交换机名称 3 路由key
chan.queueBind("fanout01", "fanoutEx", "");
chan.basicQos(1);
//定义消费者
QueueingConsumer consumer=new QueueingConsumer(chan);
//消费者与队列绑定
chan.basicConsume("fanout01",false, consumer);
while(true){
Delivery delivery= consumer.nextDelivery();
System.out.println("一号消费者接收到"+
new String(delivery.getBody()));
chan.basicAck(delivery.getEnvelope().
getDeliveryTag(), false);
}
}
@Test
public void receiv02() throws Exception{
//获取连接
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
//生命队列
chan.queueDeclare("fanout02", false, false, false, null);
//声明交换机
chan.exchangeDeclare("fanoutEx", "fanout");
//绑定队列到交换机
//参数 1 队列名称,2 交换机名称 3 路由key
chan.queueBind("fanout02", "fanoutEx", "");
chan.basicQos(1);
//定义消费者
QueueingConsumer consumer=new QueueingConsumer(chan);
//消费者与队列绑定
chan.basicConsume("fanout02",false, consumer);
while(true){
Delivery delivery= consumer.nextDelivery();
System.out.println("二号消费者接收到"+new String(delivery.getBody()));
chan.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
}4 routing路由模式
生产者发送消息到交换机,同时绑定一个路由Key,交换机根据路由key对下游绑定的队列进行路
由key的判断,满足路由key的队列才会接收到消息,消费者消费消息
应用场景: 项目中的error报错
案例：
public class RoutingTopicTest {
@Test
public void routingSend() throws Exception{
//获取连接
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
//声明交换机
//参数意义,1 交换机名称,2 类型:fanout,direct,topic
chan.exchangeDeclare("directEx", "direct");
//发送消息
String msg="路由模式的消息";
chan.basicPublish("directEx", "jt1713",
null, msg.getBytes());
}
@Test
public void routingRec01() throws Exception{
System.out.println("一号消费者等待接收消息");
//获取连接
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
//声明队列
chan.queueDeclare("direct01", false, false, false, null);
//声明交换机
chan.exchangeDeclare("directEx", "direct");
//绑定队列到交换机
//参数 1 队列名称,2 交换机名称 3 路由key
chan.queueBind("direct01", "directEx", "jt1712");
chan.basicQos(1);
//定义消费者
QueueingConsumer consumer=new QueueingConsumer(chan);
//消费者与队列绑定
chan.basicConsume("direct01",false, consumer);
while(true){
Delivery delivery= consumer.nextDelivery();
System.out.println("一号消费者接收到"+
new String(delivery.getBody()));
chan.basicAck(delivery.getEnvelope().
getDeliveryTag(), false);
}
}
@Test
public void routingRec02() throws Exception{
System.out.println("二号消费者等待接收消息");
//获取连接
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
//声明队列
chan.queueDeclare("direct02", false, false, false, null);
//声明交换机
chan.exchangeDeclare("directEx", "direct");
//绑定队列到交换机
//参数 1 队列名称,2 交换机名称 3 路由key
chan.queueBind("direct02", "directEx", "jt1711");
chan.basicQos(1);
//定义消费者
QueueingConsumer consumer=new QueueingConsumer(chan);
//消费者与队列绑定
chan.basicConsume("direct02",false, consumer);
while(true){
Delivery delivery= consumer.nextDelivery();
System.out.println("二号消费者接收到"+
new String(delivery.getBody()));
chan.basicAck(delivery.getEnvelope().
getDeliveryTag(), false);
}
}
}5 topic主题模式
*号代表单个词语
#代表多个词语
其他的内容与routing路由模式一致
案例：
public class RoutingTopicTest {
@Test
public void routingRec02() throws Exception{
System.out.println("二号消费者等待接收消息");
//获取连接
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
//声明队列
chan.queueDeclare("direct02", false, false, false, null);
//声明交换机
chan.exchangeDeclare("directEx", "direct");
//绑定队列到交换机
//参数 1 队列名称,2 交换机名称 3 路由key
chan.queueBind("direct02", "directEx", "jt1711");
chan.basicQos(1);
//定义消费者
QueueingConsumer consumer=new QueueingConsumer(chan);
//消费者与队列绑定
chan.basicConsume("direct02",false, consumer);
while(true){
Delivery delivery= consumer.nextDelivery();
System.out.println("二号消费者接收到"+
new String(delivery.getBody()));
chan.basicAck(delivery.getEnvelope().
getDeliveryTag(), false);
}
}
@Test
public void topicSend() throws Exception{
//获取连接
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
//声明交换机
//参数意义,1 交换机名称,2 类型:fanout,direct,topic
chan.exchangeDeclare("topicEx", "topic");
//发送消息
String msg="主题模式的消息";
chan.basicPublish("topicEx", "jt1712.add.update",
null, msg.getBytes());
}
@Test
public void topicRec01() throws Exception{
System.out.println("一号消费者等待接收消息");
//获取连接
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
//声明队列
chan.queueDeclare("topic01", false, false, false, null);
//声明交换机
chan.exchangeDeclare("topicEx", "topic");
//绑定队列到交换机
//参数 1 队列名称,2 交换机名称 3 路由key
chan.queueBind("topic01", "topicEx", "jt1712");
chan.basicQos(1);
//定义消费者
QueueingConsumer consumer=new QueueingConsumer(chan);
//消费者与队列绑定
chan.basicConsume("topic01",false, consumer);
while(true){
Delivery delivery= consumer.nextDelivery();
System.out.println("一号消费者接收到"+
new String(delivery.getBody()));
chan.basicAck(delivery.getEnvelope().
getDeliveryTag(), false);
}
}
@Test
public void topicRec02() throws Exception{
System.out.println("二号消费者等待接收消息");
//获取连接
Connection conn = ConnectionUtil.getConn();
Channel chan = conn.createChannel();
//声明队列
chan.queueDeclare("topic02", false, false, false, null);
//声明交换机
chan.exchangeDeclare("topicEx", "topic");
//绑定队列到交换机
//参数 1 队列名称,2 交换机名称 3 路由key
chan.queueBind("topic02", "topicEx", "jt1712.#");
chan.basicQos(1);
//定义消费者
QueueingConsumer consumer=new QueueingConsumer(chan);
//消费者与队列绑定
chan.basicConsume("topic02",false, consumer);
while(true){
Delivery delivery= consumer.nextDelivery();
System.out.println("二号消费者接收到"+
new String(delivery.getBody()));
chan.basicAck(delivery.getEnvelope().
getDeliveryTag(), false);
}
}
}智能推荐
oracle 12c 集群安装后的检查_12c查看crs状态-程序员宅基地
文章浏览阅读1.6k次。安装配置gi、安装数据库软件、dbca建库见下：http://blog.csdn.net/kadwf123/article/details/784299611、检查集群节点及状态：[root@rac2 ~]# olsnodes -srac1 Activerac2 Activerac3 Activerac4 Active[root@rac2 ~]_12c查看crs状态
解决jupyter notebook无法找到虚拟环境的问题_jupyter没有pytorch环境-程序员宅基地
文章浏览阅读1.3w次，点赞45次，收藏99次。我个人用的是anaconda3的一个python集成环境，自带jupyter notebook，但在我打开jupyter notebook界面后，却找不到对应的虚拟环境，原来是jupyter notebook只是通用于下载anaconda时自带的环境，其他环境要想使用必须手动下载一些库：1.首先进入到自己创建的虚拟环境(pytorch是虚拟环境的名字)activate pytorch2.在该环境下下载这个库conda install ipykernelconda install nb__jupyter没有pytorch环境
国内安装scoop的保姆教程_scoop-cn-程序员宅基地
文章浏览阅读5.2k次，点赞19次，收藏28次。选择scoop纯属意外，也是无奈，因为电脑用户被锁了管理员权限，所有exe安装程序都无法安装，只可以用绿色软件，最后被我发现scoop，省去了到处下载XXX绿色版的烦恼，当然scoop里需要管理员权限的软件也跟我无缘了（譬如everything）。推荐添加dorado这个bucket镜像，里面很多中文软件，但是部分国外的软件下载地址在github，可能无法下载。以上两个是官方bucket的国内镜像，所有软件建议优先从这里下载。上面可以看到很多bucket以及软件数。如果官网登陆不了可以试一下以下方式。_scoop-cn
Element ui colorpicker在Vue中的使用_vue el-color-picker-程序员宅基地
文章浏览阅读4.5k次，点赞2次，收藏3次。首先要有一个color-picker组件 <el-color-picker v-model="headcolor"></el-color-picker>在data里面data() { return {headcolor: ’ #278add ’ //这里可以选择一个默认的颜色} }然后在你想要改变颜色的地方用v-bind绑定就好了，例如：这里的:sty..._vue el-color-picker
迅为iTOP-4412精英版之烧写内核移植后的镜像_exynos 4412 刷机-程序员宅基地
文章浏览阅读640次。基于芯片日益增长的问题，所以内核开发者们引入了新的方法，就是在内核中只保留函数，而数据则不包含，由用户（应用程序员）自己把数据按照规定的格式编写，并放在约定的地方，为了不占用过多的内存，还要求数据以根精简的方式编写。boot启动时，传参给内核，告诉内核设备树文件和kernel的位置，内核启动时根据地址去找到设备树文件，再利用专用的编译器去反编译dtb文件，将dtb还原成数据结构，以供驱动的函数去调用。firmware是三星的一个固件的设备信息，因为找不到固件，所以内核启动不成功。_exynos 4412 刷机
Linux系统配置jdk_linux配置jdk-程序员宅基地
文章浏览阅读2w次，点赞24次，收藏42次。Linux系统配置jdkLinux学习教程，Linux入门教程（超详细）_linux配置jdk
随便推点
matlab(4)：特殊符号的输入_matlab微米怎么输入-程序员宅基地
文章浏览阅读3.3k次，点赞5次，收藏19次。xlabel('\delta');ylabel('AUC');具体符号的对照表参照下图：_matlab微米怎么输入
C语言程序设计-文件(打开与关闭、顺序、二进制读写)-程序员宅基地
文章浏览阅读119次。顺序读写指的是按照文件中数据的顺序进行读取或写入。对于文本文件，可以使用fgets、fputs、fscanf、fprintf等函数进行顺序读写。在C语言中，对文件的操作通常涉及文件的打开、读写以及关闭。文件的打开使用fopen函数，而关闭则使用fclose函数。在C语言中，可以使用fread和fwrite函数进行二进制读写。 Biaoge 于2024-03-09 23:51发布 阅读量：7 ️文章类型：【 C语言程序设计 】在C语言中，用于打开文件的函数是____，用于关闭文件的函数是____。
Touchdesigner自学笔记之三_touchdesigner怎么让一个模型跟着鼠标移动-程序员宅基地
文章浏览阅读3.4k次，点赞2次，收藏13次。跟随鼠标移动的粒子以grid（SOP）为partical（SOP）的资源模板，调整后连接【Geo组合+point spirit（MAT)】，在连接【feedback组合】适当调整。影响粒子动态的节点【metaball(SOP)+force(SOP)】添加mouse in（CHOP)鼠标位置到metaball的坐标，实现鼠标影响。..._touchdesigner怎么让一个模型跟着鼠标移动
【附源码】基于java的校园停车场管理系统的设计与实现61m0e9计算机毕设SSM_基于java技术的停车场管理系统实现与设计-程序员宅基地
文章浏览阅读178次。项目运行环境配置：Jdk1.8 + Tomcat7.0 + Mysql + HBuilderX（Webstorm也行）+ Eclispe（IntelliJ IDEA,Eclispe,MyEclispe,Sts都支持）。项目技术：Springboot + mybatis + Maven +mysql5.7或8.0+html+css+js等等组成，B/S模式 + Maven管理等等。环境需要1.运行环境：最好是java jdk 1.8，我们在这个平台上运行的。其他版本理论上也可以。_基于java技术的停车场管理系统实现与设计
Android系统播放器MediaPlayer源码分析_android多媒体播放源码分析 时序图-程序员宅基地
文章浏览阅读3.5k次。前言对于MediaPlayer播放器的源码分析内容相对来说比较多，会从Java-&amp;gt;Jni-&amp;gt;C/C++慢慢分析，后面会慢慢更新。另外，博客只作为自己学习记录的一种方式，对于其他的不过多的评论。MediaPlayerDemopublic class MainActivity extends AppCompatActivity implements SurfaceHolder.Cal..._android多媒体播放源码分析 时序图
java 数据结构与算法 ——快速排序法-程序员宅基地
文章浏览阅读2.4k次，点赞41次，收藏13次。java 数据结构与算法 ——快速排序法_快速排序法