rabbitmq版本和pom依赖问题
重点：一定要注意rabbitmq的版本和pom依赖的版本
----我的rabbitmq版本是3.6.10，pom依赖版本是
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.0.4</version>
</dependency>
不然会出现以下错误
G:\develop\software\java\bin\java.exe "-javaagent:G:\develop\software\idea\IntelliJ IDEA 2020.1\lib\idea_rt.jar=53792:G:\develop\software\idea\IntelliJ IDEA 2020.1\bin" -Dfile.encoding=UTF-8 -classpath G:\develop\software\java\jre\lib\charsets.jar;G:\develop\software\java\jre\lib\deploy.jar;G:\develop\software\java\jre\lib\ext\access-bridge-64.jar;G:\develop\software\java\jre\lib\ext\cldrdata.jar;G:\develop\software\java\jre\lib\ext\dnsns.jar;G:\develop\software\java\jre\lib\ext\jaccess.jar;G:\develop\software\java\jre\lib\ext\jfxrt.jar;G:\develop\software\java\jre\lib\ext\localedata.jar;G:\develop\software\java\jre\lib\ext\nashorn.jar;G:\develop\software\java\jre\lib\ext\sunec.jar;G:\develop\software\java\jre\lib\ext\sunjce_provider.jar;G:\develop\software\java\jre\lib\ext\sunmscapi.jar;G:\develop\software\java\jre\lib\ext\sunpkcs11.jar;G:\develop\software\java\jre\lib\ext\zipfs.jar;G:\develop\software\java\jre\lib\javaws.jar;G:\develop\software\java\jre\lib\jce.jar;G:\develop\software\java\jre\lib\jfr.jar;G:\develop\software\java\jre\lib\jfxswt.jar;G:\develop\software\java\jre\lib\jsse.jar;G:\develop\software\java\jre\lib\management-agent.jar;G:\develop\software\java\jre\lib\plugin.jar;G:\develop\software\java\jre\lib\resources.jar;G:\develop\software\java\jre\lib\rt.jar;D:\xiangmu\ideaxiangmu\ideaxiangmu\xcEduService01\test-rabbitmq-producer\target\test-classes;D:\xiangmu\ideaxiangmu\ideaxiangmu\xcEduService01\test-rabbitmq-producer\target\classes;D:\software\develop\maven\repository_xczx\org\springframework\boot\spring-boot-starter-amqp\1.4.5.RELEASE\spring-boot-starter-amqp-1.4.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\boot\spring-boot-starter\2.0.1.RELEASE\spring-boot-starter-2.0.1.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\boot\spring-boot\2.0.1.RELEASE\spring-boot-2.0.1.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\boot\spring-boot-autoconfigure\2.0.1.RELEASE\spring-boot-autoconfigure-2.0.1.RELEASE.jar;D:\software\develop\maven\repository_xczx\javax\annotation\javax.annotation-api\1.3.2\javax.annotation-api-1.3.2.jar;D:\software\develop\maven\repository_xczx\org\yaml\snakeyaml\1.19\snakeyaml-1.19.jar;D:\software\develop\maven\repository_xczx\org\springframework\spring-messaging\5.0.5.RELEASE\spring-messaging-5.0.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\spring-beans\5.0.5.RELEASE\spring-beans-5.0.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\amqp\spring-rabbit\2.0.3.RELEASE\spring-rabbit-2.0.3.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\amqp\spring-amqp\2.0.3.RELEASE\spring-amqp-2.0.3.RELEASE.jar;D:\software\develop\maven\repository_xczx\com\rabbitmq\amqp-client\5.1.2\amqp-client-5.1.2.jar;D:\software\develop\maven\repository_xczx\com\rabbitmq\http-client\2.0.1.RELEASE\http-client-2.0.1.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\spring-webflux\5.0.5.RELEASE\spring-webflux-5.0.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\io\projectreactor\reactor-core\3.1.6.RELEASE\reactor-core-3.1.6.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\reactivestreams\reactive-streams\1.0.2\reactive-streams-1.0.2.jar;D:\software\develop\maven\repository_xczx\org\apache\httpcomponents\httpclient\4.5.5\httpclient-4.5.5.jar;D:\software\develop\maven\repository_xczx\org\apache\httpcomponents\httpcore\4.4.9\httpcore-4.4.9.jar;D:\software\develop\maven\repository_xczx\commons-codec\commons-codec\1.10\commons-codec-1.10.jar;D:\software\develop\maven\repository_xczx\com\fasterxml\jackson\core\jackson-databind\2.9.5\jackson-databind-2.9.5.jar;D:\software\develop\maven\repository_xczx\com\fasterxml\jackson\core\jackson-annotations\2.9.0\jackson-annotations-2.9.0.jar;D:\software\develop\maven\repository_xczx\com\fasterxml\jackson\core\jackson-core\2.9.5\jackson-core-2.9.5.jar;D:\software\develop\maven\repository_xczx\io\projectreactor\ipc\reactor-netty\0.7.6.RELEASE\reactor-netty-0.7.6.RELEASE.jar;D:\software\develop\maven\repository_xczx\io\netty\netty-codec-http\4.1.23.Final\netty-codec-http-4.1.23.Final.jar;D:\software\develop\maven\repository_xczx\io\netty\netty-codec\4.1.23.Final\netty-codec-4.1.23.Final.jar;D:\software\develop\maven\repository_xczx\io\netty\netty-handler\4.1.23.Final\netty-handler-4.1.23.Final.jar;D:\software\develop\maven\repository_xczx\io\netty\netty-buffer\4.1.23.Final\netty-buffer-4.1.23.Final.jar;D:\software\develop\maven\repository_xczx\io\netty\netty-transport\4.1.23.Final\netty-transport-4.1.23.Final.jar;D:\software\develop\maven\repository_xczx\io\netty\netty-resolver\4.1.23.Final\netty-resolver-4.1.23.Final.jar;D:\software\develop\maven\repository_xczx\io\netty\netty-handler-proxy\4.1.23.Final\netty-handler-proxy-4.1.23.Final.jar;D:\software\develop\maven\repository_xczx\io\netty\netty-codec-socks\4.1.23.Final\netty-codec-socks-4.1.23.Final.jar;D:\software\develop\maven\repository_xczx\io\netty\netty-transport-native-epoll\4.1.23.Final\netty-transport-native-epoll-4.1.23.Final.jar;D:\software\develop\maven\repository_xczx\io\netty\netty-common\4.1.23.Final\netty-common-4.1.23.Final.jar;D:\software\develop\maven\repository_xczx\io\netty\netty-transport-native-unix-common\4.1.23.Final\netty-transport-native-unix-common-4.1.23.Final.jar;D:\software\develop\maven\repository_xczx\org\springframework\spring-context\5.0.5.RELEASE\spring-context-5.0.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\spring-aop\5.0.5.RELEASE\spring-aop-5.0.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\spring-expression\5.0.5.RELEASE\spring-expression-5.0.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\spring-tx\5.0.5.RELEASE\spring-tx-5.0.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\spring-web\5.0.5.RELEASE\spring-web-5.0.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\retry\spring-retry\1.2.2.RELEASE\spring-retry-1.2.2.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\boot\spring-boot-starter-test\2.0.1.RELEASE\spring-boot-starter-test-2.0.1.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\boot\spring-boot-test\2.0.1.RELEASE\spring-boot-test-2.0.1.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\boot\spring-boot-test-autoconfigure\2.0.1.RELEASE\spring-boot-test-autoconfigure-2.0.1.RELEASE.jar;D:\software\develop\maven\repository_xczx\com\jayway\jsonpath\json-path\2.4.0\json-path-2.4.0.jar;D:\software\develop\maven\repository_xczx\net\minidev\json-smart\2.3\json-smart-2.3.jar;D:\software\develop\maven\repository_xczx\net\minidev\accessors-smart\1.2\accessors-smart-1.2.jar;D:\software\develop\maven\repository_xczx\org\ow2\asm\asm\5.0.4\asm-5.0.4.jar;D:\software\develop\maven\repository_xczx\org\slf4j\slf4j-api\1.7.25\slf4j-api-1.7.25.jar;D:\software\develop\maven\repository_xczx\junit\junit\4.12\junit-4.12.jar;D:\software\develop\maven\repository_xczx\org\assertj\assertj-core\3.9.1\assertj-core-3.9.1.jar;D:\software\develop\maven\repository_xczx\org\mockito\mockito-core\2.15.0\mockito-core-2.15.0.jar;D:\software\develop\maven\repository_xczx\net\bytebuddy\byte-buddy\1.7.11\byte-buddy-1.7.11.jar;D:\software\develop\maven\repository_xczx\net\bytebuddy\byte-buddy-agent\1.7.11\byte-buddy-agent-1.7.11.jar;D:\software\develop\maven\repository_xczx\org\objenesis\objenesis\2.6\objenesis-2.6.jar;D:\software\develop\maven\repository_xczx\org\hamcrest\hamcrest-core\1.3\hamcrest-core-1.3.jar;D:\software\develop\maven\repository_xczx\org\hamcrest\hamcrest-library\1.3\hamcrest-library-1.3.jar;D:\software\develop\maven\repository_xczx\org\skyscreamer\jsonassert\1.5.0\jsonassert-1.5.0.jar;D:\software\develop\maven\repository_xczx\com\vaadin\external\google\android-json\0.0.20131108.vaadin1\android-json-0.0.20131108.vaadin1.jar;D:\software\develop\maven\repository_xczx\org\springframework\spring-core\5.0.5.RELEASE\spring-core-5.0.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\spring-jcl\5.0.5.RELEASE\spring-jcl-5.0.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\springframework\spring-test\5.0.5.RELEASE\spring-test-5.0.5.RELEASE.jar;D:\software\develop\maven\repository_xczx\org\xmlunit\xmlunit-core\2.5.1\xmlunit-core-2.5.1.jar;D:\software\develop\maven\repository_xczx\org\springframework\boot\spring-boot-starter-logging\2.0.1.RELEASE\spring-boot-starter-logging-2.0.1.RELEASE.jar;D:\software\develop\maven\repository_xczx\ch\qos\logback\logback-classic\1.2.3\logback-classic-1.2.3.jar;D:\software\develop\maven\repository_xczx\ch\qos\logback\logback-core\1.2.3\logback-core-1.2.3.jar;D:\software\develop\maven\repository_xczx\org\apache\logging\log4j\log4j-to-slf4j\2.10.0\log4j-to-slf4j-2.10.0.jar;D:\software\develop\maven\repository_xczx\org\apache\logging\log4j\log4j-api\2.10.0\log4j-api-2.10.0.jar;D:\software\develop\maven\repository_xczx\org\slf4j\jul-to-slf4j\1.7.25\jul-to-slf4j-1.7.25.jar com.xuecheng.test.rabbitmq.Producer02_publish
java.util.concurrent.TimeoutException
at com.rabbitmq.utility.BlockingCell.get(BlockingCell.java:77)
at com.rabbitmq.utility.BlockingCell.uninterruptibleGet(BlockingCell.java:120)
at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:36)
at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:494)
at com.rabbitmq.client.impl.AMQConnection.start(AMQConnection.java:306)
at com.rabbitmq.client.impl.recovery.RecoveryAwareAMQConnectionFactory.newConnection(RecoveryAwareAMQConnectionFactory.java:64)
at com.rabbitmq.client.impl.recovery.AutorecoveringConnection.init(AutorecoveringConnection.java:99)
at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:948)
at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:907)
at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:865)
at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:1018)
at com.xuecheng.test.rabbitmq.Producer02_publish.main(Producer02_publish.java:32)
23:52:07.755 [AMQP Connection 192.168.80.132:5672] ERROR com.rabbitmq.client.impl.ForgivingExceptionHandler - An unexpected connection driver error occured
java.net.SocketException: Socket Closed
at java.net.SocketInputStream.socketRead0(Native Method)
at java.net.SocketInputStream.socketRead(SocketInputStream.java:116)
at java.net.SocketInputStream.read(SocketInputStream.java:171)
at java.net.SocketInputStream.read(SocketInputStream.java:141)
at java.io.BufferedInputStream.fill(BufferedInputStream.java:246)
at java.io.BufferedInputStream.read(BufferedInputStream.java:265)
at java.io.DataInputStream.readUnsignedByte(DataInputStream.java:288)
at com.rabbitmq.client.impl.Frame.readFrom(Frame.java:91)
at com.rabbitmq.client.impl.SocketFrameHandler.readFrame(SocketFrameHandler.java:164)
at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:580)
at java.lang.Thread.run(Thread.java:748)
Process finished with exit code 0
pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.1.RELEASE</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.cgdata</groupId>
<artifactId>firsttest</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>firsttest</name>
<description>Demo project for Spring Boot</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>5.3.10</version>
</dependency>
<dependency>
<groupId>org.apache.poi</groupId>
<artifactId>poi-ooxml</artifactId>
<version>3.17</version>
</dependency>
<dependency>
<groupId>com.oracle</groupId>
<artifactId>ojdbc6</artifactId>
<version>11.2.0.3</version>
</dependency>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>3.0.4</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.junit.vintage</groupId>
<artifactId>junit-vintage-engine</artifactId>
</exclusion>
</exclusions>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
生产者
package com.cgdata.testrabbitmq;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.concurrent.TimeoutException;
public class Producer02_publish {
//队列名称
private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
//交换机名称
private static final String EXCHANGE_FANOUT_INFORM = "exchange_fanout_inform";
public static void main(String[] args) throws IOException, TimeoutException {
Connection connection = null;
Channel channel = null;
try {
//通过连接工厂创建新的连接和mq进行连接
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.80.132"); //设置mq所在服务器的ip
factory.setPort(5672); //设置mq所在服务器的端口号
factory.setUsername("admin");//设置登录用户名
factory.setPassword("admin");//设置登录密码
factory.setVirtualHost("/");//rabbitmq默认虚拟机名称为“/”，相当于一个独立的mq服务
//创建与RabbitMQ服务的TCP连接
connection = factory.newConnection();
//创建与Exchange(交换机)的通道，每个连接可以创建多个通道，每个通道代表一个会话任务
channel = connection.createChannel();
/**
* //声明交换机 String exchange, BuiltinExchangeType type
* 参数明细
* 1、交换机名称
* 2、交换机类型，fanout、topic、direct、headers
*/
channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, "fanout");
/**
* 声明队列，如果Rabbit中没有此队列将自动创建
* param1:队列名称
* param2:是否持久化
* param3:队列是否独占此连接
* param4:队列不再使用时是否自动删除此队列
* param5:队列参数(可以设置一个队列的扩展参数，比如：可设置存活时间)
*/
channel.queueDeclare(QUEUE_INFORM_EMAIL, true, false, false, null);
channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
/**
* //交换机和队列绑定String queue, String exchange, String routingKey
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key
*/
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,"");
channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_FANOUT_INFORM,"");
//发送消息
for (int i = 0; i < 5; i++) {
String message = "helloword" + System.currentTimeMillis();
StringBuilder message1 = new StringBuilder("helloword1++");
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMdd HH:mm:dd");
String date = simpleDateFormat.format(System.currentTimeMillis());
message1.append(date.substring(8, 14));
// message1.append(date.substring(8,date.indexOf(":")));
// message1.append(date.substring(8,date.lastIndexOf(":")));
//message1.append(date.split(":"));
/**
* 消息发布方法
* param1：Exchange的名称，如果没有指定，则使用Default Exchange
* param2:routingKey,消息的路由Key，是用于Exchange（交换机）将消息转发到指定的消息队列
* param3:消息包含的属性
* param4：消息体
*/
/**
* 这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显
示绑定或解除绑定
* 默认的交换机，routingKey等于队列名称
*/
channel.basicPublish(EXCHANGE_FANOUT_INFORM,"", null, message1.toString().getBytes());
System.out.println("Send message is:'" + message1 + "'");
}
} catch (Exception e) {
e.printStackTrace();
} finally {
if (channel != null) {
channel.close();
}
if (connection != null) {
connection.close();
}
}
}
}
消费者email
package com.cgdata.testrabbitmq;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer02_subscribe_email {
//队列名称 queue_inform_sms queue_inform_email
private static final String QUEUE_INFORM_EMAIL = "queue_inform_email";
//交换机名称
private static final String EXCHANGE_FANOUT_INFORM = "enchange_fanout_inform";
public static void main(String[] args) {
Connection connection = null;
Channel channel = null;
try {
//创建连接工厂用于获取连接
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.80.132");//设置主机
factory.setPort(5672);//设置端口号
factory.setUsername("admin");//登录用户名
factory.setPassword("admin");//登录密码
factory.setVirtualHost("/");//设置mq虚拟机，一个虚拟机相当于一个独立的mq
//创建连接
connection = factory.newConnection();
//创建与交换机的通道，每个通道代表一个会话
channel = connection.createChannel();
/**
* //声明交换机 String exchange, BuiltinExchangeType type
* 参数明细
* 1、交换机名称
* 2、交换机类型，fanout、topic、direct、headers
*/
channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM,"fanout");//BuiltinExchangeType.FANOUT
//声明队列
// channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
/**
* 参数明细：
* 1、队列名称
* 2、是否持久化
* 3、是否独占此队列
* 4、队列不用是否自动删除
* 5、参数
*/
channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
//交换机和队列绑定String queue, String exchange, String routingKey
/**
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key
*/
channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,"");
//定义消费方法
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
* 消费者接收消息调用此方法
* @param consumerTag 消费者的标签，在channel.basicConsume()去指定
* @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
(收到消息失败后是否需要重新发送)
* @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();
//路由
String routingKey = envelope.getRoutingKey();
//消息id
long deliveryTag = envelope.getDeliveryTag();
//消息内容
String msg = new String(body,"utf-8");
System.out.println("receive message.."+ msg);
}
};
/**
* 监听队列String queue, boolean autoAck,Consumer callback
* 参数明细
* 1、队列名称
* 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置
为false则需要手动回复
* 3、消费消息的方法，消费者接收到消息后调用此方法
*/
channel.basicConsume(QUEUE_INFORM_EMAIL,true,consumer);
} catch (IOException e) {
e.printStackTrace();
}
}
}
消费者sms
package com.cgdata.testrabbitmq;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.TimeoutException;
public class Consumer02_subscribe_sms {
//队列名称
private static final String QUEUE_INFORM_SMS = "queue_inform_sms";
//交换机名称
private static final String EXCHANGE_FANOUT_INFORM = "enchange_fanout_inform";
public static void main(String[] args) {
Connection connection = null;
Channel channel = null;
try {
//创建连接工厂用于获取连接
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.80.132");//设置主机
factory.setPort(5672);//设置端口号
factory.setUsername("admin");//登录用户名
factory.setPassword("admin");//登录密码
factory.setVirtualHost("/");//设置mq虚拟机，一个虚拟机相当于一个独立的mq
//创建连接
connection = factory.newConnection();
//创建与交换机的通道，每个通道代表一个会话
channel = connection.createChannel();
/**
* //声明交换机 String exchange, BuiltinExchangeType type
* 参数明细
* 1、交换机名称
* 2、交换机类型，fanout、topic、direct、headers
*/
channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM,"fanout");//BuiltinExchangeType.FANOUT
//声明队列
// channel.queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
/**
* 参数明细：
* 1、队列名称
* 2、是否持久化
* 3、是否独占此队列
* 4、队列不用是否自动删除
* 5、参数
*/
channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
//交换机和队列绑定String queue, String exchange, String routingKey
/**
* 参数明细
* 1、队列名称
* 2、交换机名称
* 3、路由key
*/
channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_FANOUT_INFORM,"");
//定义消费方法
DefaultConsumer consumer = new DefaultConsumer(channel){
/**
* 消费者接收消息调用此方法
* @param consumerTag 消费者的标签，在channel.basicConsume()去指定
* @param envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志
(收到消息失败后是否需要重新发送)
* @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();
//路由
String routingKey = envelope.getRoutingKey();
//消息id
long deliveryTag = envelope.getDeliveryTag();
//消息内容
String msg = new String(body,"utf-8");
System.out.println("receive message.."+ msg);
}
};
/**
* 监听队列String queue, boolean autoAck,Consumer callback
* 参数明细
* 1、队列名称
* 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置
为false则需要手动回复
* 3、消费消息的方法，消费者接收到消息后调用此方法
*/
channel.basicConsume(QUEUE_INFORM_SMS,true,consumer);
} catch (IOException e) {
e.printStackTrace();
}
}
}
/**
* 监听队列String queue, boolean autoAck,Consumer callback
* 参数明细
* 1、队列名称
* 2、是否自动回复，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置
为false则需要手动回复
* 3、消费消息的方法，消费者接收到消息后调用此方法
*/
channel.basicConsume(QUEUE_INFORM_SMS,true,consumer);
} catch (IOException e) {
e.printStackTrace();
}
}
}
版权声明：本文为qq_24790827原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接和本声明。