在普通maven项目应用中使用MQ

tech2023-07-31 104

新人学习，本文从B站学习借鉴而来，链接：https://www.bilibili.com/video/BV14A411q7pF/?p=9

本文会介绍maven项目创建，通过项目介绍简单模式Simple、工作模式Work、订阅模式Publish/Subscribe、路由模式Routing这四种常用的模式

在普通maven项目应用中使用MQ

maven项目创建步骤：1.简单模式1.1.消息生产者1.2.消息消费者 2.工作模式2.1.消息生产者2.2.消息消费者12.3.消息消费者2 3.订阅模式3.1.消息生产者3.2.消息消费者13.3.消息消费者2 4.路由模式4.1.消息生产者4.2.消息消费者14.3.消息消费者2 结语：

maven项目创建步骤：

1.首先，创建maven项目。

这里我创建了聚合项目（也可以不聚合，producer和consumer分开放也行，这里我是为了方便打包），该聚合项目包含producer和consumer，根据模式不同，消费者consumer会有1到多个，继续往下就知道了。

2.然后，配置pom.xml文件（添加依赖）、日志文件。

四种模式的生产者和消费者都要配置pom.xml和日志，值得注意的是日志是一样的可以直接复制过去，但pom.xml文件头部信息不同，把依赖部分添加即可，不要把整个pom.xml直接拷贝过去，否则会出现一些bug。

3.编写类，实现发送接收消息功能。

分别在producer和consumer中，写发送消息的类sendMsg和接受消息的类receiveMsg，这两个类是主要类。

还有一个工具类ConnectionUtil，用于连接rabbitmq的，连接是一样的所以producer和consumer中，这个工具类除了头部包名不同，具体内容一样，复制的时候注意头部别一起复制了。

以简单模式为例，一个生产者一个消费者，配置后目录：

producer/consumer的pom.xml的依赖部分: _{simple父项目的pom.xml不用添加这部分依赖}

<dependencies>  <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.5.0</version> </dependency>  <dependency> <groupId>org.slf4j</groupId> <artifactId>slf4j-log4j12</artifactId> <version>1.7.25</version> <scope>test</scope> </dependency>  <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-lang3</artifactId> <version>3.9</version> </dependency> </dependencies>

在src/main/resources目录下，添加log4j.properties日志：

log4j.properties：

log4j.rootLogger=DEBUG,A1 log4j.logger.com.taotao = DEBUG log4j.logger.org.mybatis = DEBUG log4j.appender.A1=org.apache.log4j.ConsoleAppender log4j.appender.A1.layout=org.apache.log4j.PatternLayout log4j.appender.A1.layout.ConversionPattern=%-d{yyyy-MM-dd HH:mm:ss,SSS} [%t] [%c]-[%p] %m%n

1.简单模式

在下图中，“ P”是我们的生产者，“ C”是我们的消费者。中间的框是一个队列-RabbitMQ代表使用者保留的消息缓冲区。生产者将消息发送到“ hello”队列。使用者从该队列接收消息。

1.1.消息生产者

首先，创建帮助类ConnectionUtil.java。这个是一个连接MQ的工具类，是为了代码简洁，所以另外创建一个连接的工具类，使用的话直接导入该包调用即可。 ConnectionUtil：

package com.rbmq.consumer.utils; import com.rabbitmq.client.Connection; import com.rabbitmq.client.ConnectionFactory; public class ConnectionUtil { public static Connection getConnection() throws Exception { //定义连接工厂 ConnectionFactory factory = new ConnectionFactory(); // 设置服务地址 factory.setHost("127.0.0.1"); //端口 factory.setPort(5672); // 设置账号信息，虚拟主机、用户名、密码，这里不特指某一个虚拟主机，所以用默认的“/” factory.setVirtualHost("/"); factory.setUsername("guest"); factory.setPassword("guest"); // 通过工程获取连接 Connection connection = factory.newConnection(); return connection; } }

生产者发送消息，SendMsg.java

package com.rbmq.producer.service; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rbmq.producer.utils.ConnectionUtil; public class sendMsg { public static void main(String[] argv) throws Exception { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); // 相当于数据库中的创建连接 // 从连接中创建通道 Channel channel = connection.createChannel(); // 相当于数据库中的 statement // 声明（创建）队列,如果存在就不创建,不存在就创建 // 参数1 队列名, // 参数2 durable：是否持久化, 队列的声明默认是存放到内存中的，如果rabbitmq重启会丢失，如果想重启之后还存在就要使队列持久化，保存到Erlang自带的Mnesia数据库中，当rabbitmq重启之后会读取该数据库 // 参数3 exclusive：是否排外的，有两个作用，一：当连接关闭时connection.close()该队列是否会自动删除；二：该队列是否是私有的private，如果不是排外的，可以使用两个消费者都访问同一个队列，没有任何问题，如果是排外的，会对当前队列加锁，其他通道channel是不能访问的，如果强制访问会报异常： com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=405, reply-text=RESOURCE_LOCKED - cannot obtain exclusive access to locked queue 'queue_name' in vhost '/', class-id=50, method-id=20)一般等于true的话用于一个队列只能有一个消费者来消费的场景 // 参数4 autoDelete：是否自动删除，当最后一个消费者断开连接之后队列是否自动被删除，可以通过RabbitMQ Management，查看某个队列的消费者数量，当consumers = 0时队列就会自动删除 // 参数5 arguments：参数 //channel.queueDeclare("queue1", false, false, true, null); // 消息内容 String message = "Hello consumer!"; // 参数1 交换机,此处无 // 参数2 发送到哪个队列 // 参数3 属性 // 参数4 内容 channel.basicPublish("", "queue1", null, message.getBytes()); // 将消息发动到数据库 System.out.println(" 发送数据： '" + message + "'"); //关闭通道和连接 channel.close(); connection.close(); } }

测试结果：

1.2.消息消费者

同样，首先创建工具类ConnectionUtil.java，这个类是连接mq的，没有连接mq，接收不到消息然后，编写接收消息类receiveMsg：

package com.rbmq.consumer.service; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.rbmq.consumer.utils.ConnectionUtil; public class receiveMsg { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); // 相当于jdbc里的数据库连接 Channel channel = connection.createChannel();// 相当于jdbc操作的statement Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag,Envelope envelope, AMQP.BasicProperties properties,byte[] body) { //body为从队列中获取的数据 String msg = new String(body); System.out.println("接收到的消息："+msg); } }; channel.basicConsume("queue1", true, consumer); } }

测试结果：

2.工作模式

工作队列，生产者把消息发送到队列中，由队列分配到消费者，采用的是先到先得的原则，即C1、C2共同争抢当前的消息队列内容,谁先拿到谁负责消费消息。例如超市低价促销先到先得。

基本项目步骤和上面的一样，都是：

#mermaid-svg-YHQRRskYJtTc3W0R .label{font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family);fill:#333;color:#333}#mermaid-svg-YHQRRskYJtTc3W0R .label text{fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R .node rect,#mermaid-svg-YHQRRskYJtTc3W0R .node circle,#mermaid-svg-YHQRRskYJtTc3W0R .node ellipse,#mermaid-svg-YHQRRskYJtTc3W0R .node polygon,#mermaid-svg-YHQRRskYJtTc3W0R .node path{fill:#ECECFF;stroke:#9370db;stroke-width:1px}#mermaid-svg-YHQRRskYJtTc3W0R .node .label{text-align:center;fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R .node.clickable{cursor:pointer}#mermaid-svg-YHQRRskYJtTc3W0R .arrowheadPath{fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R .edgePath .path{stroke:#333;stroke-width:1.5px}#mermaid-svg-YHQRRskYJtTc3W0R .flowchart-link{stroke:#333;fill:none}#mermaid-svg-YHQRRskYJtTc3W0R .edgeLabel{background-color:#e8e8e8;text-align:center}#mermaid-svg-YHQRRskYJtTc3W0R .edgeLabel rect{opacity:0.9}#mermaid-svg-YHQRRskYJtTc3W0R .edgeLabel span{color:#333}#mermaid-svg-YHQRRskYJtTc3W0R .cluster rect{fill:#ffffde;stroke:#aa3;stroke-width:1px}#mermaid-svg-YHQRRskYJtTc3W0R .cluster text{fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family);font-size:12px;background:#ffffde;border:1px solid #aa3;border-radius:2px;pointer-events:none;z-index:100}#mermaid-svg-YHQRRskYJtTc3W0R .actor{stroke:#ccf;fill:#ECECFF}#mermaid-svg-YHQRRskYJtTc3W0R text.actor>tspan{fill:#000;stroke:none}#mermaid-svg-YHQRRskYJtTc3W0R .actor-line{stroke:grey}#mermaid-svg-YHQRRskYJtTc3W0R .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333}#mermaid-svg-YHQRRskYJtTc3W0R .messageLine1{stroke-width:1.5;stroke-dasharray:2, 2;stroke:#333}#mermaid-svg-YHQRRskYJtTc3W0R #arrowhead path{fill:#333;stroke:#333}#mermaid-svg-YHQRRskYJtTc3W0R .sequenceNumber{fill:#fff}#mermaid-svg-YHQRRskYJtTc3W0R #sequencenumber{fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R #crosshead path{fill:#333;stroke:#333}#mermaid-svg-YHQRRskYJtTc3W0R .messageText{fill:#333;stroke:#333}#mermaid-svg-YHQRRskYJtTc3W0R .labelBox{stroke:#ccf;fill:#ECECFF}#mermaid-svg-YHQRRskYJtTc3W0R .labelText,#mermaid-svg-YHQRRskYJtTc3W0R .labelText>tspan{fill:#000;stroke:none}#mermaid-svg-YHQRRskYJtTc3W0R .loopText,#mermaid-svg-YHQRRskYJtTc3W0R .loopText>tspan{fill:#000;stroke:none}#mermaid-svg-YHQRRskYJtTc3W0R .loopLine{stroke-width:2px;stroke-dasharray:2, 2;stroke:#ccf;fill:#ccf}#mermaid-svg-YHQRRskYJtTc3W0R .note{stroke:#aa3;fill:#fff5ad}#mermaid-svg-YHQRRskYJtTc3W0R .noteText,#mermaid-svg-YHQRRskYJtTc3W0R .noteText>tspan{fill:#000;stroke:none}#mermaid-svg-YHQRRskYJtTc3W0R .activation0{fill:#f4f4f4;stroke:#666}#mermaid-svg-YHQRRskYJtTc3W0R .activation1{fill:#f4f4f4;stroke:#666}#mermaid-svg-YHQRRskYJtTc3W0R .activation2{fill:#f4f4f4;stroke:#666}#mermaid-svg-YHQRRskYJtTc3W0R .mermaid-main-font{font-family:"trebuchet ms", verdana, arial;font-family:var(--mermaid-font-family)}#mermaid-svg-YHQRRskYJtTc3W0R .section{stroke:none;opacity:0.2}#mermaid-svg-YHQRRskYJtTc3W0R .section0{fill:rgba(102,102,255,0.49)}#mermaid-svg-YHQRRskYJtTc3W0R .section2{fill:#fff400}#mermaid-svg-YHQRRskYJtTc3W0R .section1,#mermaid-svg-YHQRRskYJtTc3W0R .section3{fill:#fff;opacity:0.2}#mermaid-svg-YHQRRskYJtTc3W0R .sectionTitle0{fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R .sectionTitle1{fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R .sectionTitle2{fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R .sectionTitle3{fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R .sectionTitle{text-anchor:start;font-size:11px;text-height:14px;font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family)}#mermaid-svg-YHQRRskYJtTc3W0R .grid .tick{stroke:#d3d3d3;opacity:0.8;shape-rendering:crispEdges}#mermaid-svg-YHQRRskYJtTc3W0R .grid .tick text{font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family)}#mermaid-svg-YHQRRskYJtTc3W0R .grid path{stroke-width:0}#mermaid-svg-YHQRRskYJtTc3W0R .today{fill:none;stroke:red;stroke-width:2px}#mermaid-svg-YHQRRskYJtTc3W0R .task{stroke-width:2}#mermaid-svg-YHQRRskYJtTc3W0R .taskText{text-anchor:middle;font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family)}#mermaid-svg-YHQRRskYJtTc3W0R .taskText:not([font-size]){font-size:11px}#mermaid-svg-YHQRRskYJtTc3W0R .taskTextOutsideRight{fill:#000;text-anchor:start;font-size:11px;font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family)}#mermaid-svg-YHQRRskYJtTc3W0R .taskTextOutsideLeft{fill:#000;text-anchor:end;font-size:11px}#mermaid-svg-YHQRRskYJtTc3W0R .task.clickable{cursor:pointer}#mermaid-svg-YHQRRskYJtTc3W0R .taskText.clickable{cursor:pointer;fill:#003163 !important;font-weight:bold}#mermaid-svg-YHQRRskYJtTc3W0R .taskTextOutsideLeft.clickable{cursor:pointer;fill:#003163 !important;font-weight:bold}#mermaid-svg-YHQRRskYJtTc3W0R .taskTextOutsideRight.clickable{cursor:pointer;fill:#003163 !important;font-weight:bold}#mermaid-svg-YHQRRskYJtTc3W0R .taskText0,#mermaid-svg-YHQRRskYJtTc3W0R .taskText1,#mermaid-svg-YHQRRskYJtTc3W0R .taskText2,#mermaid-svg-YHQRRskYJtTc3W0R .taskText3{fill:#fff}#mermaid-svg-YHQRRskYJtTc3W0R .task0,#mermaid-svg-YHQRRskYJtTc3W0R .task1,#mermaid-svg-YHQRRskYJtTc3W0R .task2,#mermaid-svg-YHQRRskYJtTc3W0R .task3{fill:#8a90dd;stroke:#534fbc}#mermaid-svg-YHQRRskYJtTc3W0R .taskTextOutside0,#mermaid-svg-YHQRRskYJtTc3W0R .taskTextOutside2{fill:#000}#mermaid-svg-YHQRRskYJtTc3W0R .taskTextOutside1,#mermaid-svg-YHQRRskYJtTc3W0R .taskTextOutside3{fill:#000}#mermaid-svg-YHQRRskYJtTc3W0R .active0,#mermaid-svg-YHQRRskYJtTc3W0R .active1,#mermaid-svg-YHQRRskYJtTc3W0R .active2,#mermaid-svg-YHQRRskYJtTc3W0R .active3{fill:#bfc7ff;stroke:#534fbc}#mermaid-svg-YHQRRskYJtTc3W0R .activeText0,#mermaid-svg-YHQRRskYJtTc3W0R .activeText1,#mermaid-svg-YHQRRskYJtTc3W0R .activeText2,#mermaid-svg-YHQRRskYJtTc3W0R .activeText3{fill:#000 !important}#mermaid-svg-YHQRRskYJtTc3W0R .done0,#mermaid-svg-YHQRRskYJtTc3W0R .done1,#mermaid-svg-YHQRRskYJtTc3W0R .done2,#mermaid-svg-YHQRRskYJtTc3W0R .done3{stroke:grey;fill:#d3d3d3;stroke-width:2}#mermaid-svg-YHQRRskYJtTc3W0R .doneText0,#mermaid-svg-YHQRRskYJtTc3W0R .doneText1,#mermaid-svg-YHQRRskYJtTc3W0R .doneText2,#mermaid-svg-YHQRRskYJtTc3W0R .doneText3{fill:#000 !important}#mermaid-svg-YHQRRskYJtTc3W0R .crit0,#mermaid-svg-YHQRRskYJtTc3W0R .crit1,#mermaid-svg-YHQRRskYJtTc3W0R .crit2,#mermaid-svg-YHQRRskYJtTc3W0R .crit3{stroke:#f88;fill:red;stroke-width:2}#mermaid-svg-YHQRRskYJtTc3W0R .activeCrit0,#mermaid-svg-YHQRRskYJtTc3W0R .activeCrit1,#mermaid-svg-YHQRRskYJtTc3W0R .activeCrit2,#mermaid-svg-YHQRRskYJtTc3W0R .activeCrit3{stroke:#f88;fill:#bfc7ff;stroke-width:2}#mermaid-svg-YHQRRskYJtTc3W0R .doneCrit0,#mermaid-svg-YHQRRskYJtTc3W0R .doneCrit1,#mermaid-svg-YHQRRskYJtTc3W0R .doneCrit2,#mermaid-svg-YHQRRskYJtTc3W0R .doneCrit3{stroke:#f88;fill:#d3d3d3;stroke-width:2;cursor:pointer;shape-rendering:crispEdges}#mermaid-svg-YHQRRskYJtTc3W0R .milestone{transform:rotate(45deg) scale(0.8, 0.8)}#mermaid-svg-YHQRRskYJtTc3W0R .milestoneText{font-style:italic}#mermaid-svg-YHQRRskYJtTc3W0R .doneCritText0,#mermaid-svg-YHQRRskYJtTc3W0R .doneCritText1,#mermaid-svg-YHQRRskYJtTc3W0R .doneCritText2,#mermaid-svg-YHQRRskYJtTc3W0R .doneCritText3{fill:#000 !important}#mermaid-svg-YHQRRskYJtTc3W0R .activeCritText0,#mermaid-svg-YHQRRskYJtTc3W0R .activeCritText1,#mermaid-svg-YHQRRskYJtTc3W0R .activeCritText2,#mermaid-svg-YHQRRskYJtTc3W0R .activeCritText3{fill:#000 !important}#mermaid-svg-YHQRRskYJtTc3W0R .titleText{text-anchor:middle;font-size:18px;fill:#000;font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family)}#mermaid-svg-YHQRRskYJtTc3W0R g.classGroup text{fill:#9370db;stroke:none;font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family);font-size:10px}#mermaid-svg-YHQRRskYJtTc3W0R g.classGroup text .title{font-weight:bolder}#mermaid-svg-YHQRRskYJtTc3W0R g.clickable{cursor:pointer}#mermaid-svg-YHQRRskYJtTc3W0R g.classGroup rect{fill:#ECECFF;stroke:#9370db}#mermaid-svg-YHQRRskYJtTc3W0R g.classGroup line{stroke:#9370db;stroke-width:1}#mermaid-svg-YHQRRskYJtTc3W0R .classLabel .box{stroke:none;stroke-width:0;fill:#ECECFF;opacity:0.5}#mermaid-svg-YHQRRskYJtTc3W0R .classLabel .label{fill:#9370db;font-size:10px}#mermaid-svg-YHQRRskYJtTc3W0R .relation{stroke:#9370db;stroke-width:1;fill:none}#mermaid-svg-YHQRRskYJtTc3W0R .dashed-line{stroke-dasharray:3}#mermaid-svg-YHQRRskYJtTc3W0R #compositionStart{fill:#9370db;stroke:#9370db;stroke-width:1}#mermaid-svg-YHQRRskYJtTc3W0R #compositionEnd{fill:#9370db;stroke:#9370db;stroke-width:1}#mermaid-svg-YHQRRskYJtTc3W0R #aggregationStart{fill:#ECECFF;stroke:#9370db;stroke-width:1}#mermaid-svg-YHQRRskYJtTc3W0R #aggregationEnd{fill:#ECECFF;stroke:#9370db;stroke-width:1}#mermaid-svg-YHQRRskYJtTc3W0R #dependencyStart{fill:#9370db;stroke:#9370db;stroke-width:1}#mermaid-svg-YHQRRskYJtTc3W0R #dependencyEnd{fill:#9370db;stroke:#9370db;stroke-width:1}#mermaid-svg-YHQRRskYJtTc3W0R #extensionStart{fill:#9370db;stroke:#9370db;stroke-width:1}#mermaid-svg-YHQRRskYJtTc3W0R #extensionEnd{fill:#9370db;stroke:#9370db;stroke-width:1}#mermaid-svg-YHQRRskYJtTc3W0R .commit-id,#mermaid-svg-YHQRRskYJtTc3W0R .commit-msg,#mermaid-svg-YHQRRskYJtTc3W0R .branch-label{fill:lightgrey;color:lightgrey;font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family)}#mermaid-svg-YHQRRskYJtTc3W0R .pieTitleText{text-anchor:middle;font-size:25px;fill:#000;font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family)}#mermaid-svg-YHQRRskYJtTc3W0R .slice{font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family)}#mermaid-svg-YHQRRskYJtTc3W0R g.stateGroup text{fill:#9370db;stroke:none;font-size:10px;font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family)}#mermaid-svg-YHQRRskYJtTc3W0R g.stateGroup text{fill:#9370db;fill:#333;stroke:none;font-size:10px}#mermaid-svg-YHQRRskYJtTc3W0R g.statediagram-cluster .cluster-label text{fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R g.stateGroup .state-title{font-weight:bolder;fill:#000}#mermaid-svg-YHQRRskYJtTc3W0R g.stateGroup rect{fill:#ECECFF;stroke:#9370db}#mermaid-svg-YHQRRskYJtTc3W0R g.stateGroup line{stroke:#9370db;stroke-width:1}#mermaid-svg-YHQRRskYJtTc3W0R .transition{stroke:#9370db;stroke-width:1;fill:none}#mermaid-svg-YHQRRskYJtTc3W0R .stateGroup .composit{fill:white;border-bottom:1px}#mermaid-svg-YHQRRskYJtTc3W0R .stateGroup .alt-composit{fill:#e0e0e0;border-bottom:1px}#mermaid-svg-YHQRRskYJtTc3W0R .state-note{stroke:#aa3;fill:#fff5ad}#mermaid-svg-YHQRRskYJtTc3W0R .state-note text{fill:black;stroke:none;font-size:10px}#mermaid-svg-YHQRRskYJtTc3W0R .stateLabel .box{stroke:none;stroke-width:0;fill:#ECECFF;opacity:0.7}#mermaid-svg-YHQRRskYJtTc3W0R .edgeLabel text{fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R .stateLabel text{fill:#000;font-size:10px;font-weight:bold;font-family:'trebuchet ms', verdana, arial;font-family:var(--mermaid-font-family)}#mermaid-svg-YHQRRskYJtTc3W0R .node circle.state-start{fill:black;stroke:black}#mermaid-svg-YHQRRskYJtTc3W0R .node circle.state-end{fill:black;stroke:white;stroke-width:1.5}#mermaid-svg-YHQRRskYJtTc3W0R #statediagram-barbEnd{fill:#9370db}#mermaid-svg-YHQRRskYJtTc3W0R .statediagram-cluster rect{fill:#ECECFF;stroke:#9370db;stroke-width:1px}#mermaid-svg-YHQRRskYJtTc3W0R .statediagram-cluster rect.outer{rx:5px;ry:5px}#mermaid-svg-YHQRRskYJtTc3W0R .statediagram-state .divider{stroke:#9370db}#mermaid-svg-YHQRRskYJtTc3W0R .statediagram-state .title-state{rx:5px;ry:5px}#mermaid-svg-YHQRRskYJtTc3W0R .statediagram-cluster.statediagram-cluster .inner{fill:white}#mermaid-svg-YHQRRskYJtTc3W0R .statediagram-cluster.statediagram-cluster-alt .inner{fill:#e0e0e0}#mermaid-svg-YHQRRskYJtTc3W0R .statediagram-cluster .inner{rx:0;ry:0}#mermaid-svg-YHQRRskYJtTc3W0R .statediagram-state rect.basic{rx:5px;ry:5px}#mermaid-svg-YHQRRskYJtTc3W0R .statediagram-state rect.divider{stroke-dasharray:10,10;fill:#efefef}#mermaid-svg-YHQRRskYJtTc3W0R .note-edge{stroke-dasharray:5}#mermaid-svg-YHQRRskYJtTc3W0R .statediagram-note rect{fill:#fff5ad;stroke:#aa3;stroke-width:1px;rx:0;ry:0}:root{--mermaid-font-family: '"trebuchet ms", verdana, arial';--mermaid-font-family: "Comic Sans MS", "Comic Sans", cursive}#mermaid-svg-YHQRRskYJtTc3W0R .error-icon{fill:#522}#mermaid-svg-YHQRRskYJtTc3W0R .error-text{fill:#522;stroke:#522}#mermaid-svg-YHQRRskYJtTc3W0R .edge-thickness-normal{stroke-width:2px}#mermaid-svg-YHQRRskYJtTc3W0R .edge-thickness-thick{stroke-width:3.5px}#mermaid-svg-YHQRRskYJtTc3W0R .edge-pattern-solid{stroke-dasharray:0}#mermaid-svg-YHQRRskYJtTc3W0R .edge-pattern-dashed{stroke-dasharray:3}#mermaid-svg-YHQRRskYJtTc3W0R .edge-pattern-dotted{stroke-dasharray:2}#mermaid-svg-YHQRRskYJtTc3W0R .marker{fill:#333}#mermaid-svg-YHQRRskYJtTc3W0R .marker.cross{stroke:#333} :root { --mermaid-font-family: "trebuchet ms", verdana, arial;} #mermaid-svg-YHQRRskYJtTc3W0R { color: rgba(0, 0, 0, 0.75); font: ; } 创建maven项目导入依赖和日志编写实现类实现

项目创建和依赖、日志、ConnectionUtil帮助类一样的，不同的是多了一个消费者：

2.1.消息生产者

sendMsg:

package com.rbmq.producer.service; import java.util.Scanner; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rbmq.producer.utils.ConnectionUtil; public class sendMsg { public static void main(String[] argv) throws Exception { System.out.println("请输入消息："); Scanner scaner = new Scanner(System.in); String msg=null; while(!"quit".equals(msg=scaner.nextLine())) { // 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); // 从连接中创建通道 Channel channel = connection.createChannel(); // 参数1 交换机,此处无 // 参数2 发送到哪个队列 // 参数3 属性 // 参数4 内容 channel.basicPublish("", "queue4", null, msg.getBytes()); // 将消息发动到数据库 System.out.println(" 发送数据： '" + msg + "'"); //关闭通道和连接 channel.close(); connection.close(); } } }

测试环节（中间有丢包情况，所以多发了一次hello msg3）：

2.2.消息消费者1

package com.rbmq.consumer.service; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.rbmq.consumer.utils.ConnectionUtil; public class receiveMsg { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); // 相当于jdbc里的数据库连接 Channel channel = connection.createChannel();// 相当于jdbc操作的statement Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag,Envelope envelope, AMQP.BasicProperties properties,byte[] body) { //body为从队列中获取的数据 String msg = new String(body); System.out.println("consumer1接收到的消息："+msg); } }; channel.basicConsume("queue4", true, consumer); } }

测试结果：

2.3.消息消费者2

package com.rbmq.consumer.service; import com.rabbitmq.client.AMQP; import com.rabbitmq.client.Channel; import com.rabbitmq.client.Connection; import com.rabbitmq.client.Consumer; import com.rabbitmq.client.DefaultConsumer; import com.rabbitmq.client.Envelope; import com.rbmq.consumer.utils.ConnectionUtil; public class receiveMsg { public static void main(String[] args) throws Exception { Connection connection = ConnectionUtil.getConnection(); // 相当于jdbc里的数据库连接 Channel channel = connection.createChannel();// 相当于jdbc操作的statement Consumer consumer = new DefaultConsumer(channel) { @Override public void handleDelivery(String consumerTag,Envelope envelope, AMQP.BasicProperties properties,byte[] body) { //body为从队列中获取的数据 String msg = new String(body); System.out.println("consumer2接收到的消息："+msg); } }; channel.basicConsume("queue4", true, consumer); } }

测试结果：

3.订阅模式

一次向许多消费者发送消息，X代表交换机rabbitMQ内部组件,消息产生者将消息放入交换机,交换机发布订阅把消息发送到所有消息队列中,对应消息队列的消费者拿到消息进行消费，例如广告。

项目和上面的基本不变，主要改变发送接收模式参数即可。

3.1.消息生产者

sendMsg代码和上面的工作模式基本没有改变，改变的只有这一行：

channel.basicPublish("ex1", "", null, msg.getBytes());

注释：把消息发送到交换机ex1上而不是队列中，而在rabbitmq管理页面上交换机ex1绑定了队列1和队列2

测试环节：

3.2.消息消费者1

同样，receiveMsg1也基本没有改变，改变的是这一行：

channel.basicConsume("queue1", true, consumer);

消息消费者1接收的是交换机ex1所绑定的queue1队列

测试结果：

3.3.消息消费者2

同样，receiveMsg2也基本没有改变，改变的是这一行：

channel.basicConsume("queue2", true, consumer);

消息消费者1接收的是交换机ex1所绑定的queue2队列

测试结果：注：这两个消费者都提前运行，同时接收到这两个消息。

4.路由模式

路由模式，消息生产者将消息发送给交换机按照路由判断,路由是字符串(info) ,交换机根据路由的key,只能匹配上路由key对应的消息队列,对应的消费者才能消费消息。项目同样没有多大改动

4.1.消息生产者

sendMsg主要代码，其余不变：

// 获取到连接以及mq通道 Connection connection = ConnectionUtil.getConnection(); // 从连接中创建通道 Channel channel = connection.createChannel(); // 参数1 交换机 // 参数2 路由key,如果前面的交换机为空，这里是队列参数 // 参数3 属性 // 参数4 内容 if(msg.startsWith("a")) { channel.basicPublish("ex2", "a", null, msg.getBytes()); }else if(msg.startsWith("b")) { channel.basicPublish("ex2", "b", null, msg.getBytes()); } // 将消息发动到数据库 System.out.println(" 发送数据： '" + msg + "'"); //关闭通道和连接 channel.close(); connection.close();

测试环节：

4.2.消息消费者1

channel.basicConsume("queue3", true, consumer);

交换机ex2绑定了queue3，路由key为a，消费者1直接监听队列3

测试结果：消费者1只接受路由key为a的消息，前面把字符串以a开头的路由key设置为a

4.3.消息消费者2

channel.basicConsume("queue4", true, consumer);

交换机ex2绑定了queue4，路由key为b，消费者2直接监听队列4

测试结果：消费者2只接受路由key为b的消息，前面把字符串以b开头的路由key设置为b

结语：

通过四个项目的实现，对rabbitmq的几种常用的模式有比较好的了解，不足的地方是，经常出现丢包现象。附上项目代码：链接：https://pan.baidu.com/s/1jv-WZv9zQ6fU_jnQQRb73Q 提取码：pkar

最新回复(0)