Java操作RabbitMQ添加队列、消费队列和三个交换机

254人浏览 / 0人评论

一、发送消息到队列(生产者)

新建一个maven项目,在pom.xml文件加入以下依赖

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

新建一个P1类

package com.rabbitMQ.test;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
/**
 * @author mowen
 * @create 2019/11/20-11:23
 */
public class P1 {
 public static void main(String[] args) throws IOException, TimeoutException {
 //消息队列名字
 String queueName="queue";
 //实例连接工厂
 ConnectionFactory connectionFactory=new ConnectionFactory();
 //设置地址
 connectionFactory.setHost("192.168.128.233");
 //设置端口
 connectionFactory.setPort(5672);
 //设置用户名
 connectionFactory.setUsername("mowen");
 //设置密码
 connectionFactory.setPassword("123456");
 //获取连接(跟jdbc很像)
 Connection connection = connectionFactory.newConnection();
 //创建通道
 Channel channel = connection.createChannel();
 //声明队列。
 //参数1:队列名
 //参数2:持久化 (true表示是,队列将在服务器重启时依旧存在)
 //参数3:独占队列(创建者可以使用的私有队列,断开后自动删除)
 //参数4:当所有消费者客户端连接断开时是否自动删除队列
 //参数5:队列的其他参数
 channel.queueDeclare(queueName,true,false,false,null);
 
 for (int i = 0; i < 10; i++) {
 String msg="msg"+i;
 // 基本发布消息
 // 第一个参数为交换机名称、
 // 第二个参数为队列映射的路由key、
 // 第三个参数为消息的其他属性、
 // 第四个参数为发送信息的主体
 channel.basicPublish("",queueName,null,msg.getBytes());
 }
 
 channel.close();
 connection.close();
 }
}

运行后再浏览器进入RabbitMQ的控制台,切换到queue看到

Java操作RabbitMQ添加队列、消费队列和三个交换机

 

二、获取队列消息(消费者)

新建一个C1类

package com.rabbitMQ.test;
 
 
import com.rabbitmq.client.*;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
/**
 * @author mowen
 * @create 2019/11/20-13:12
 */
public class C1 {
 public static void main(String[] args) throws IOException, TimeoutException {
 //消息队列名字
 String queueName="queue";
 //实例连接工厂
 ConnectionFactory connectionFactory=new ConnectionFactory();
 //设置地址
 connectionFactory.setHost("192.168.128.233");
 //设置端口
 connectionFactory.setPort(5672);
 //设置用户名
 connectionFactory.setUsername("mowen");
 //设置密码
 connectionFactory.setPassword("123456");
 //获取连接(跟jdbc很像)
 Connection connection = connectionFactory.newConnection();
 //创建通道
 Channel channel = connection.createChannel();
 
 // 创建一个消费者
 Consumer consumer = new DefaultConsumer(channel){
 @Override
 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
 // 消费收到消息的时候调用的回调
 System.out.println("C3接收到:" + new String(body));
 }
 };
 
 
 //把消费着绑定到指定队列
 //第一个是队列名
 //第二个是 是否自动确认
 //第三个是消费者
 channel.basicConsume(queueName,true,consumer);
 
 }
}

运行后输出为

Java操作RabbitMQ添加队列、消费队列和三个交换机

 


消费者一般都不会关闭,会一直等待队列消息,可以手动关闭程序。

channel.basicConsume(queueName,true,consumer);中的true为收到消息后自动确认,改为false取消自动确认。

在handleDelivery方法最后面用

channel.basicAck(envelope.getDeliveryTag(),false);

来收到手动确认消息。消费者可以有多个并且可以同时消费一个队列;

当有多个消费者同时消费同一个队列时,收到的消息是平均分配的(消费者没收到之前已经确认每个消费者受到的消息),

但当其中一个消费者性能差的话,会影响其他的消费者,因为还要等它收完消息,这样会拖累其他消费者。

可以设置channel 的basicQos方法

//设置最多接受消息数量
// 设置了这个参数之后要吧自动确认关掉
channel.basicQos(1);

三、扇形(fanout)交换机

扇形交换机是基本的交换机类型,会把收到的消息以广播的形式发送到绑定的队列里,因为不需要经过条件筛选,所以它的速度最快。

在生产者项目新建一个fanout类

package com.rabbitMQ.routing;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
/**
 * @author mowen
 * @date 2019/11/20-11:23
 */
public class fanout {
 public static void main(String[] args) throws IOException, TimeoutException {
 //交换机名字
 String exchangeName="fanout";
 //交换机名字类型
 String exchangeType="fanout";
 //消息队列名字
 String queueName1="fanout.queue1";
 String queueName2="fanout.queue2";
 String queueName3="fanout.queue3";
 //实例连接工厂
 ConnectionFactory connectionFactory=new ConnectionFactory();
 //设置地址
 connectionFactory.setHost("192.168.128.233");
 //设置端口
 connectionFactory.setPort(5672);
 //设置用户名
 connectionFactory.setUsername("mowen");
 //设置密码
 connectionFactory.setPassword("123456");
 //获取连接(跟jdbc很像)
 Connection connection = connectionFactory.newConnection();
 //创建通道
 Channel channel = connection.createChannel();
 //声明队列。
 //参数1:队列名
 //参数2:持久化 (true表示是,队列将在服务器重启时依旧存在)
 //参数3:独占队列(创建者可以使用的私有队列,断开后自动删除)
 //参数4:当所有消费者客户端连接断开时是否自动删除队列
 //参数5:队列的其他参数
 channel.queueDeclare(queueName1,true,false,false,null);
 channel.queueDeclare(queueName2,true,false,false,null);
 channel.queueDeclare(queueName3,true,false,false,null);
 
 //声明交换机
 channel.exchangeDeclare(exchangeName,exchangeType);
 
 //队列绑定到交换机
 channel.queueBind(queueName1,exchangeName,"");
 channel.queueBind(queueName2,exchangeName,"");
 channel.queueBind(queueName3,exchangeName,"");
 
 for (int i = 0; i < 10; i++) {
 String msg="msg"+i;
 // 基本发布消息
 // 第一个参数为交换机名称、
 // 第二个参数为队列映射的路由key、
 // 第三个参数为消息的其他属性、
 // 第四个参数为发送信息的主体
 channel.basicPublish(exchangeName,"",null,msg.getBytes());
 }
 
 channel.close();
 connection.close();
 }
}

运行后在RabbitMQ网页管理后台的queue会看到

Java操作RabbitMQ添加队列、消费队列和三个交换机

 

切换到Exchanges会看到一个

Java操作RabbitMQ添加队列、消费队列和三个交换机

 

就是我们声明的交换机,点击会看到我们绑定的队列

Java操作RabbitMQ添加队列、消费队列和三个交换机

 

四、直连(direct)交换机

直连交换机会带路由功能,队列通过routing_key与直连交换机绑定,发送消息需要指定routing_key,交换机收到消息时,交换机会根据routing_key发送到指定队列里,同样的routing_key可以支持多个队列。

在生产者项目新建direct类

package com.rabbitMQ.routing;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
/**
 * @author mowen
 * @date 2019/11/20-11:23
 */
public class direct {
 public static void main(String[] args) throws IOException, TimeoutException {
 String exchangeName="direct";
 String exchangeType="direct";
 //消息队列名字
 String queueName1="direct.queue1";
 String queueName2="direct.queue2";
 String queueName3="direct.queue3";
 //实例连接工厂
 ConnectionFactory connectionFactory=new ConnectionFactory();
 //设置地址
 connectionFactory.setHost("192.168.128.233");
 //设置端口
 connectionFactory.setPort(5672);
 //设置用户名
 connectionFactory.setUsername("mowen");
 //设置密码
 connectionFactory.setPassword("123456");
 //获取连接(跟jdbc很像)
 Connection connection = connectionFactory.newConnection();
 //创建通道
 Channel channel = connection.createChannel();
 //声明队列。
 //参数1:队列名
 //参数2:持久化 (true表示是,队列将在服务器重启时依旧存在)
 //参数3:独占队列(创建者可以使用的私有队列,断开后自动删除)
 //参数4:当所有消费者客户端连接断开时是否自动删除队列
 //参数5:队列的其他参数
 channel.queueDeclare(queueName1,true,false,false,null);
 channel.queueDeclare(queueName2,true,false,false,null);
 channel.queueDeclare(queueName3,true,false,false,null);
 
 //声明交换机
 channel.exchangeDeclare(exchangeName,exchangeType);
 
 //队列绑定到交换机并指定rouing_key
 channel.queueBind(queueName1,exchangeName,"key1");
 channel.queueBind(queueName2,exchangeName,"key2");
 channel.queueBind(queueName3,exchangeName,"key1");
 
 for (int i = 0; i < 10; i++) {
 String msg="msg"+i;
 // 基本发布消息
 // 第一个参数为交换机名称、
 // 第二个参数为队列映射的路由key、
 // 第三个参数为消息的其他属性、
 // 第四个参数为发送信息的主体
 channel.basicPublish(exchangeName,"key1",null,msg.getBytes());
 }
 
 channel.close();
 connection.close();
 }
}

运行后到后台的queue会看到

Java操作RabbitMQ添加队列、消费队列和三个交换机

 

切换到Exchanges会看到

Java操作RabbitMQ添加队列、消费队列和三个交换机

 

点击进去

Java操作RabbitMQ添加队列、消费队列和三个交换机

 

五、主题(topic)交换机

主题交换机的routing_key可以有一定的规则,交换机和队列的routing_key需要采用.#.…..的格式

每个部分用.分开

*代表一个单词(不是字符)

代表任意数量(0或n个)单词

在生产者项目新进topic类

package com.rabbitMQ.routing;
 
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
 
import java.io.IOException;
import java.util.concurrent.TimeoutException;
 
/**
 * @author mowen
 * @date 2019/11/20-11:23
 */
public class topic {
 public static void main(String[] args) throws IOException, TimeoutException {
 String exchangeName="topic";
 String exchangeType="topic";
 //消息队列名字
 String queueName1="topic.queue1";
 String queueName2="topic.queue2";
 String queueName3="topic.queue3";
 //实例连接工厂
 ConnectionFactory connectionFactory=new ConnectionFactory();
 //设置地址
 connectionFactory.setHost("192.168.128.233");
 //设置端口
 connectionFactory.setPort(5672);
 //设置用户名
 connectionFactory.setUsername("mowen");
 //设置密码
 connectionFactory.setPassword("123456");
 //获取连接(跟jdbc很像)
 Connection connection = connectionFactory.newConnection();
 //创建通道
 Channel channel = connection.createChannel();
 //声明队列。
 //参数1:队列名
 //参数2:持久化 (true表示是,队列将在服务器重启时依旧存在)
 //参数3:独占队列(创建者可以使用的私有队列,断开后自动删除)
 //参数4:当所有消费者客户端连接断开时是否自动删除队列
 //参数5:队列的其他参数
 channel.queueDeclare(queueName1,true,false,false,null);
 channel.queueDeclare(queueName2,true,false,false,null);
 channel.queueDeclare(queueName3,true,false,false,null);
 
 //声明交换机
 channel.exchangeDeclare(exchangeName,exchangeType);
 
 //队列绑定到交换机并指定rouing_key
 channel.queueBind(queueName1,exchangeName,"com.aaa.*");
 channel.queueBind(queueName2,exchangeName,"com.*.topic");
 channel.queueBind(queueName3,exchangeName,"com.bbb.*");
 
 for (int i = 0; i < 10; i++) {
 String msg="msg"+i;
 // 基本发布消息
 // 第一个参数为交换机名称、
 // 第二个参数为队列映射的路由key、
 // 第三个参数为消息的其他属性、
 // 第四个参数为发送信息的主体
 channel.basicPublish(exchangeName,"com.aaa.topic",null,msg.getBytes());
 }
 
 channel.close();
 connection.close();
 }
}

运行后,到后台queue会看到

Java操作RabbitMQ添加队列、消费队列和三个交换机

 

切换到Exchanges会看到

Java操作RabbitMQ添加队列、消费队列和三个交换机

 

点击进入会看到

Java操作RabbitMQ添加队列、消费队列和三个交换机

转载自:https://www.toutiao.com/a6761751298875326987

全部评论