docker run -d -p 5672:5672 -p 15672:15672 --name rabbitmq rabbitmq:management

测试

基本的流程（没有交换机）
图解：

快速入门

创建一个springboot项目： rabbitmq-api
导入pom依赖

<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>3.6.5</version>
</dependency>

新建类

producer
作用：发送消息

connection：通道 channel：会话

package com.wxm.rabbitapi.quickstart;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
/*** @author wxm* @site www.wxm.com* @company xxx公司* @create 2020-02-27 22:00** 作用：发送消息**/
public class Producer {public static void main(String[] args) throws Exception {//1 创建一个ConnectionFactory, 并进行配置ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.192.138");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 通过连接工厂创建连接Connection connection = connectionFactory.newConnection();//3 通过connection创建一个ChannelChannel channel = connection.createChannel();//4 通过Channel发送数据for(int i=0; i < 5; i++){String msg = "Hello RabbitMQ!";//1 exchange   2 routingKeychannel.basicPublish("", "test001", null, msg.getBytes());}//5 记得要关闭相关的连接channel.close();connection.close();}}

consumer
作用：接受消息

package com.wxm.rabbitapi.quickstart;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;/*** @author wxm* @site www.wxm.com* @company xxx公司* @create 2020-02-27 22:00*/
public class Consumer {public static void main(String[] args) throws Exception {//1 创建一个ConnectionFactory, 并进行配置ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.192.138");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 通过连接工厂创建连接Connection connection = connectionFactory.newConnection();//3 通过connection创建一个ChannelChannel channel = connection.createChannel();//4 声明（创建）一个队列String queueName = "test001";
//        参数：队列名称、持久化与否、独占与否、无消息队列是否自动删除、消息参数
//        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)channel.queueDeclare(queueName, true, false, false, null);//5 创建消费者QueueingConsumer queueingConsumer = new QueueingConsumer(channel);//6 设置Channel
//         参数：队列名称、自动签收、消费者回调
//        basicConsume(String queue, boolean autoAck, Consumer callback)channel.basicConsume(queueName, true, queueingConsumer);while(true){//7 获取消息（Delivery：传送）QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();String msg = new String(delivery.getBody());System.err.println("消费端: " + msg);//Envelope envelope = delivery.getEnvelope();}}
}

测试：先开消费者，再开生产者
因为接收端（消费端）创建test001

生产者和消费者都要改

效果

交换机

Name：交换机名称
Type：交换机类型 direct、topic、fanout、headers
Durability：是否需要持久化，true为持久化
Auto Delete：当最后一个绑定到Exchange上的队列删除后，自动删除该Exchange
Internal：当前Exchange是否用于RabbitMQ内部使用，默认为False
Arguments：扩展参数，用于扩展AMQP协议，定制化使用

直流交换机Direct Exchange

所有发送到Direct Exchange的消息会被转发到RouteKey中指定的Queue

测试直流交换机

消费端代码

package com.wxm.rabbitapi.exchange.redirect;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
/*** @author wxm* @site www.wxm.com* @company xxx公司* @create 2020-02-28 11:33*/
public class Consumer4DirectExchange{public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory() ;connectionFactory.setHost("192.168.192.138");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setAutomaticRecoveryEnabled(true);connectionFactory.setNetworkRecoveryInterval(3000);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//4 声明String exchangeName = "test_direct_exchange";String exchangeType = "direct";String queueName = "test_direct_queue";String routingKey = "test.direct";//表示声明了一个交换机channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);//表示声明了一个队列channel.queueDeclare(queueName, false, false, false, null);//建立一个绑定关系:channel.queueBind(queueName, exchangeName, routingKey);//durable 是否持久化消息QueueingConsumer consumer = new QueueingConsumer(channel);//参数：队列名称、是否自动ACK、Consumerchannel.basicConsume(queueName, true, consumer);//循环获取消息while(true){//获取消息，如果没有消息，这一步将会一直阻塞QueueingConsumer.Delivery delivery = consumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println("收到消息：" + msg);}}
}

生产端代码

package com.wxm.rabbitapi.exchange.direct;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer4DirectExchange {public static void main(String[] args) throws Exception {//1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.192.138");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel channel = connection.createChannel();//4 声明String exchangeName = "test_direct_exchange";String routingKey = "test.direct";
//        String routingKey = "test.direct111"; //收不到//5 发送String msg = "Hello World RabbitMQ 4  Direct Exchange Message 111 ... ";channel.basicPublish(exchangeName, routingKey , null , msg.getBytes());}
}

也是先开消费者再开生产者

主题交换机

如图：大致描述过程

消费端代码

package com.wxm.rabbitapi.exchange.topic;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;public class Consumer4TopicExchange {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory() ;connectionFactory.setHost("192.168.192.138");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setAutomaticRecoveryEnabled(true);connectionFactory.setNetworkRecoveryInterval(3000);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//4 声明String exchangeName = "test_topic_exchange";String exchangeType = "topic";String queueName = "test_topic_queue";String routingKey = "user.#";
//        String routingKey = "user.*";// 1 声明交换机channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);// 2 声明队列channel.queueDeclare(queueName, false, false, false, null);// 3 建立交换机和队列的绑定关系:channel.queueBind(queueName, exchangeName, routingKey);//durable 是否持久化消息QueueingConsumer consumer = new QueueingConsumer(channel);//参数：队列名称、是否自动ACK、Consumerchannel.basicConsume(queueName, true, consumer);//循环获取消息while(true){//获取消息，如果没有消息，这一步将会一直阻塞QueueingConsumer.Delivery delivery = consumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println("收到消息：" + msg);}}
}

生产端代码

package com.wxm.rabbitapi.exchange.topic;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer4TopicExchange {public static void main(String[] args) throws Exception {//1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.192.138");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel channel = connection.createChannel();//4 声明String exchangeName = "test_topic_exchange";String routingKey1 = "user.save";String routingKey2 = "user.update";String routingKey3 = "user.delete.abc";//5 发送String msg = "Hello World RabbitMQ 4 Topic Exchange Message ...";channel.basicPublish(exchangeName, routingKey1 , null , msg.getBytes());channel.basicPublish(exchangeName, routingKey2 , null , msg.getBytes());channel.basicPublish(exchangeName, routingKey3 , null , msg.getBytes());channel.close();connection.close();}
}

输出交换机

不处理路由键，只需要简单的将队列绑定到交换机上；
发送到交换机的消息都会被转发到与该交换机绑定的所有队列上；
Fanout交换机转发消息是最快的

对比：直流交换机

Consumer4FanoutExchange

package com.wxm.rabbitapi.exchange.fanout;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;public class Consumer4FanoutExchange {public static void main(String[] args) throws Exception {ConnectionFactory connectionFactory = new ConnectionFactory() ;connectionFactory.setHost("192.168.192.138");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");connectionFactory.setAutomaticRecoveryEnabled(true);connectionFactory.setNetworkRecoveryInterval(3000);Connection connection = connectionFactory.newConnection();Channel channel = connection.createChannel();//4 声明String exchangeName = "test_fanout_exchange";String exchangeType = "fanout";String queueName = "test_fanout_queue";String routingKey = "";    //不设置路由键channel.exchangeDeclare(exchangeName, exchangeType, true, false, false, null);channel.queueDeclare(queueName, false, false, false, null);channel.queueBind(queueName, exchangeName, routingKey);//durable 是否持久化消息QueueingConsumer consumer = new QueueingConsumer(channel);//参数：队列名称、是否自动ACK、Consumerchannel.basicConsume(queueName, true, consumer);//循环获取消息while(true){//获取消息，如果没有消息，这一步将会一直阻塞QueueingConsumer.Delivery delivery = consumer.nextDelivery();String msg = new String(delivery.getBody());System.out.println("收到消息：" + msg);}}
}

生产端代码

package com.wxm.rabbitapi.exchange.fanout;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer4FanoutExchange {public static void main(String[] args) throws Exception {//1 创建ConnectionFactoryConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost("192.168.192.138");connectionFactory.setPort(5672);connectionFactory.setVirtualHost("/");//2 创建ConnectionConnection connection = connectionFactory.newConnection();//3 创建ChannelChannel channel = connection.createChannel();//4 声明String exchangeName = "test_fanout_exchange";//5 发送for(int i = 0; i < 10; i ++) {String msg = "Hello World RabbitMQ 4 FANOUT Exchange Message ...";channel.basicPublish(exchangeName, "", null , msg.getBytes());}channel.close();connection.close();}
}

== 小结：==
RabbitMQ的概念、安装与使用、管控台操作；
结合RabbitMQ的特性、Exchange、Queue、Binding、RoutingKey、Message进行核心API的讲解