RabbitMQ--基础使用(笔记)

布满荆棘的人生 2023-07-11 05:35 12阅读 0赞

pom

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

**组成说明：**

*  ConnectionFactory: 获取连接工厂
 *  Connection: 一个连接
 *  Channel: 数据通信信道，发送和接收数据
 *  Queue: 具体的消息存储队列
 *  Producer & Consumer 生产者和消费者

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4_size_16_color_FFFFFF_t_70][]

**属性：**

*  Name: 交换机名称
 *  Type: 类型direct(直连)、topic(条件连)、fanout(广播)、headers
 *  Durability: 是否需要持久化，Durable是，Transient否
 *  Auto Delete: 当最后一个绑到Exchange上的队列删除后，自动删除该Exchange
 *  Internal: 当前Exchange是否用于RabbitMQ内部使用，默认false
 *  Arguments:扩展参数，用于扩展AMQP协议自制化使用

使用 direct时，可以不指定Exchange 默认使用Exchange:default Exchange。但是RouteKey = Queue(RouteKey)

### direct ###

Producer

public static void main(String[] args) throws Exception {
    		//1 创建ConnectionFactory
    		ConnectionFactory connectionFactory = new ConnectionFactory();
    		connectionFactory.setHost("192.168.11.76");
    		connectionFactory.setPort(5672);
    		connectionFactory.setVirtualHost("/");
    		
    		//2 创建Connection
    		Connection connection = connectionFactory.newConnection();
    		//3 创建Channel
    		Channel channel = connection.createChannel();  
    		//4 声明
    		String exchangeName = "test_direct_exchange";
    		String routingKey = "test.direct111";
    		//5 发送
    		String msg = "Hello World RabbitMQ 4  Direct Exchange Message 111 ... ";
    		channel.basicPublish(exchangeName, routingKey , null , msg.getBytes());
    	}

Consumer

public static void main(String[] args) throws Exception {
            ConnectionFactory connectionFactory = new ConnectionFactory() ;
            connectionFactory.setHost("192.168.11.76");
            connectionFactory.setPort(5672);
    		connectionFactory.setVirtualHost("/");
    	//断开重连
            connectionFactory.setAutomaticRecoveryEnabled(true);
            //断开后每3秒重连一次
            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、Consumer
            channel.basicConsume(queueName, true, consumer);  
            //循环获取消息  
            while(true){  
                //获取消息，如果没有消息，这一步将会一直阻塞  
                Delivery delivery = consumer.nextDelivery();  
                String msg = new String(delivery.getBody());    
                System.out.println("收到消息：" + msg);  
            } 
    	}

### topic ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4_size_16_color_FFFFFF_t_70 1][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4_size_16_color_FFFFFF_t_70 2][]

**Producer**

public static void main(String[] args) throws Exception {
    		//1 创建ConnectionFactory
    		ConnectionFactory connectionFactory = new ConnectionFactory();
    		connectionFactory.setHost("192.168.11.76");
    		connectionFactory.setPort(5672);
    		connectionFactory.setVirtualHost("/");
    		
    		//2 创建Connection
    		Connection connection = connectionFactory.newConnection();
    		//3 创建Channel
    		Channel 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();  
    	}

Consumer

public static void main(String[] args) throws Exception {
            ConnectionFactory connectionFactory = new ConnectionFactory() ;  
            connectionFactory.setHost("192.168.11.76");
            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、Consumer
            channel.basicConsume(queueName, true, consumer);  
            //循环获取消息  
            while(true){  
                //获取消息，如果没有消息，这一步将会一直阻塞  
                Delivery delivery = consumer.nextDelivery();  
                String msg = new String(delivery.getBody());    
                System.out.println("收到消息：" + msg);  
            } 
    	}

### fanout ###

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

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4_size_16_color_FFFFFF_t_70 3][]

**Producer**

public static void main(String[] args) throws Exception {
    		//1 创建ConnectionFactory
    		ConnectionFactory connectionFactory = new ConnectionFactory();
    		connectionFactory.setHost("192.168.11.76");
    		connectionFactory.setPort(5672);
    		connectionFactory.setVirtualHost("/");
    		
    		//2 创建Connection
    		Connection connection = connectionFactory.newConnection();
    		//3 创建Channel
    		Channel 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();  
    	}

Consumer

public static void main(String[] args) throws Exception {
            ConnectionFactory connectionFactory = new ConnectionFactory() ;  
            connectionFactory.setHost("192.168.11.76");
            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、Consumer
            channel.basicConsume(queueName, true, consumer); 
            //循环获取消息  
            while(true){  
                //获取消息，如果没有消息，这一步将会一直阻塞  
                Delivery delivery = consumer.nextDelivery();  
                String msg = new String(delivery.getBody());    
                System.out.println("收到消息：" + msg);  
            } 
    	}

### Message ###

*  服务器和应用程序之间传递的数据
 *  由Properties和Payload（Body）组成
 *  常用属性：delivery mode、headers（自定义属性）
 *  其他属性：
    
     *  content\_type、content\_encoding、priority
     *  correlation\_id、reply\_to、expiration、message\_id
     *  timestamp、type、user\_id、app\_id、cluster\_id

**Producer**

public static void main(String[] args) throws Exception {
    		//1 创建一个ConnectionFactory, 并进行配置
    		ConnectionFactory connectionFactory = new ConnectionFactory();
    		connectionFactory.setHost("192.168.11.76");
    		connectionFactory.setPort(5672);
    		connectionFactory.setVirtualHost("/");
    		
    		//2 通过连接工厂创建连接
    		Connection connection = connectionFactory.newConnection();
    		//3 通过connection创建一个Channel
    		Channel channel = connection.createChannel();
    		Map<String, Object> headers = new HashMap<>();
    		headers.put("my1", "111");
    		headers.put("my2", "222");
    		AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
    				.deliveryMode(2)
    				.contentEncoding("UTF-8")
    				.expiration("10000")
    				.headers(headers)
    				.build();
    		
    		//4 通过Channel发送数据
    		for(int i=0; i < 5; i++){
    			String msg = "Hello RabbitMQ!";
    			//1 exchange   2 routingKey
    			channel.basicPublish("", "test001", properties, msg.getBytes());
    		}
    
    		//5 记得要关闭相关的连接
    		channel.close();
    		connection.close();
    	}

Consumer

public static void main(String[] args) throws Exception {
    		//1 创建一个ConnectionFactory, 并进行配置
    		ConnectionFactory connectionFactory = new ConnectionFactory();
    		connectionFactory.setHost("192.168.11.76");
    		connectionFactory.setPort(5672);
    		connectionFactory.setVirtualHost("/");
    		//2 通过连接工厂创建连接
    		Connection connection = connectionFactory.newConnection();
    		//3 通过connection创建一个Channel
    		Channel channel = connection.createChannel();
    		//4 声明（创建）一个队列
    		String queueName = "test001";
    		channel.queueDeclare(queueName, true, false, false, null);
    		//5 创建消费者
    		QueueingConsumer queueingConsumer = new QueueingConsumer(channel);
    		//6 设置Channel
    		channel.basicConsume(queueName, true, queueingConsumer);
    		while(true){
    			//7 获取消息
    			Delivery delivery = queueingConsumer.nextDelivery();
    			String msg = new String(delivery.getBody());
    			System.err.println("消费端: " + msg);
    			Map<String, Object> headers = delivery.getProperties().getHeaders();
    			System.err.println("headers get my1 value: " + headers.get("my1"));
    			
    			//Envelope envelope = delivery.getEnvelope();
    		}
    		
    	}

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200303221316813.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200303223831539.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4_size_16_color_FFFFFF_t_70 2]: https://img-blog.csdnimg.cn/20200303223851554.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4_size_16_color_FFFFFF_t_70 3]: https://img-blog.csdnimg.cn/20200303224120835.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDE4NTE4,size_16,color_FFFFFF,t_70