RabbitMQ教程_4 Java 使用rabbitmq

﹏ヽ暗。殇╰゛Y 2023-01-03 14:08 220阅读 0赞

https://gitee.com/fakerlove/rabbitmq

### 文章目录 ###

*  4. Java 使用rabbitmq
 *   *  4.1 直连模型--Helloword
     *   *  引入依赖
         *  创建开发生产者
         *  发布成功
         *  建立消费者
         *  查看是否被消费
         *  工具类
         *  使用测试
         *  项目结构
     *  4.2 work quene 任务模型
     *   *  创建生产者
         *  创建消费者
         *  消息自动确认机制
     *  4.3 fanout 模型
     *   *  开发生产者
         *  开发消费者
     *  4.4 Routing
     *   *  4.4.1 直连
         *   *  开发生产者
             *  开发消费者
             *  检验
         *  4.4.2 Routing 之订阅模型-Topic
         *   *   *  
             *  创建生产者
             *  创建消费者
             *  检查

# 4. Java 使用rabbitmq #

## 4.1 直连模型–Helloword ##

![image-20191126165840602][]

![2cbbf0faa6ffe671193b778b87db7238.png][]

### 引入依赖 ###

<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>org.example</groupId>
        <artifactId>helloword</artifactId>
        <version>1.0-SNAPSHOT</version>
    
        <dependencies>
            
            <dependency>
                <groupId>com.rabbitmq</groupId>
                <artifactId>amqp-client</artifactId>
                <version>5.10.0</version>
            </dependency>
    
        </dependencies>
    
    
    </project>

### 创建开发生产者 ###

package com.ak.test;
    
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    import org.junit.Test;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class Provider { 
    
        @Test
        public  void testSendMessage(){ 
            ConnectionFactory connectionFactory=new ConnectionFactory();
            // 设置主机名
            connectionFactory.setHost("47.100.104.187");
            // 设置端口号
            connectionFactory.setPort(5672);
            // 设置连接的虚拟主机的名字
            connectionFactory.setVirtualHost("/joker");
            // 设置虚拟机的用户名和密码
            connectionFactory.setUsername("joker");
            connectionFactory.setPassword("123456");
    
            // 获取连接对象 生产者----> 队列
    
            try { 
                // 获取连接对象
                Connection connection=connectionFactory.newConnection();
                // 获取连接中通道
                Channel channel=connection.createChannel();
                // 通道绑定对应消息队列
                /** * 参数一：队列名字，队列不存在自动创建 * 参数二，是否持久化 * 参数三：是否独占队列 true 是独占队列 ，false 不独占 * 参数四: 是否在消费完成后删除队列 * 参数五：额外附加参数 */
                channel.queueDeclare("hello",false,false,false,null);
                // 发布消息
                /** * 参数一：交换机名称 * 参数二：队列名称 * 参数三: 传递消息额外设置 * 参数四：消息的内容 * */
                channel.basicPublish("","hello",null,"hello rabbit".getBytes());
                channel.close();
                connection.close();
    
            } catch (IOException e) { 
                e.printStackTrace();
            } catch (TimeoutException e) { 
                e.printStackTrace();
            }finally { 
    
            }
    
    
        }
    }

### 发布成功 ###

![2703f2f8be2dc38a8ebcffb288a7ff58.png][]

### 建立消费者 ###

package com.ak.test;
    
    import com.rabbitmq.client.*;
    import org.junit.Test;
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class MyConsumer { 
    
        public static void main(String[] args) { 
            ConnectionFactory connectionFactory=new ConnectionFactory();
            // 设置主机名
            connectionFactory.setHost("47.100.104.187");
            // 设置端口号
            connectionFactory.setPort(5672);
            // 设置连接的虚拟主机的名字
            connectionFactory.setVirtualHost("/joker");
            // 设置虚拟机的用户名和密码
            connectionFactory.setUsername("joker");
            connectionFactory.setPassword("123456");
    
            // 获取连接对象 生产者----> 队列
    
            try { 
                // 获取连接对象
                Connection connection=connectionFactory.newConnection();
                // 获取连接中通道
                Channel channel=connection.createChannel();
                // 通道绑定对应消息队列
                /** * 参数一：队列名字，队列不存在自动创建 * 参数二，是否持久化 * 参数三：是否独占队列 true 是独占队列 ，false 不独占 * 参数四: 是否在消费完成后删除队列 * 参数五：额外附加参数 */
                channel.queueDeclare("hello",false,false,false,null);
    
                // 发布消息
                /** * 参数一：队列名称 * 参数二：开始消费的自动确认机制 * 参数三: 消费时的回调接口 * */
                channel.basicConsume("",true, new DefaultConsumer(channel){ 
    
    
                    /** * 参数回调 * @param consumerTag * @param envelope * @param properties * @param body 消息队列中取出的消息 * @throws IOException */
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println(new String(body));
                    }
                });
    
                channel.close();
                connection.close();
    
            } catch (IOException e) { 
                e.printStackTrace();
            } catch (TimeoutException e) { 
                e.printStackTrace();
            }finally { 
    
            }
        }
    }

### 查看是否被消费 ###

![dacd115c3644d07a25658673816eb2be.png][]

### 工具类 ###

package com.ak.utils;
    
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    
    import java.io.IOException;
    import java.util.concurrent.TimeoutException;
    
    public class RabbitMQUtils { 
        private static ConnectionFactory connectionFactory;
        static { 
            connectionFactory=new ConnectionFactory();
            connectionFactory.setHost("47.100.104.187");
            // 设置端口号
            connectionFactory.setPort(5672);
            // 设置连接的虚拟主机的名字
            connectionFactory.setVirtualHost("/joker");
            // 设置虚拟机的用户名和密码
            connectionFactory.setUsername("joker");
            connectionFactory.setPassword("123456");
        }
        public static Connection  getConnection(){ 
           // ConnectionFactory connectionFactory=new ConnectionFactory();
            // 设置主机名
            try { 
                // 获取连接对象
                Connection connection=connectionFactory.newConnection();
                return connection;
            } catch (IOException e) { 
                e.printStackTrace();
            } catch (TimeoutException e) { 
                e.printStackTrace();
            }finally { 
    
            }
    
            return  null;
        }
    
        public static void closeConnectionAndChanel(Channel  channel,Connection connection){ 
            try { 
                if(channel!=null){ 
                    channel.close();
                }
                if(connection!=null){ 
                    connection.close();
                }
            }catch (Exception e){ 
                e.printStackTrace();
            }
        }
    }

### 使用测试 ###

package com.ak;
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.ConnectionFactory;
    
    import java.io.IOException;
    
    public class Provider { 
        public static void main(String[] args) { 
    
            // 获取连接对象 生产者----> 队列
            try { 
                Connection connection= RabbitMQUtils.getConnection();
                Channel channel=connection.createChannel();
                channel.queueDeclare("hello",false,false,false,null);
                channel.basicPublish("","hello",null,"hello rabbit".getBytes());
                RabbitMQUtils.closeConnectionAndChanel(channel,connection);
            } catch (IOException e) { 
                e.printStackTrace();
            } finally { 
    
            }
    
        }
    }

### 项目结构 ###

![f45ecc984db1fb9f5e8fed23023783a8.png][]

## 4.2 work quene 任务模型 ##

> `Work queues`，也被称为（`Task queues`），任务模型。当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用work 模型：**让多个消费者绑定到一个队列，共同消费队列中的消息**。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。

![image-20200314221002008][]

角色：

*  P：生产者：任务的发布者
 *  C1：消费者-1，领取任务并且完成任务，假设完成速度较慢
 *  C2：消费者-2：领取任务并完成任务，假设完成速度快

### 创建生产者 ###

package com.ak.demo_2;
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    
    public class Provider { 
        public static void main(String[] args) { 
            Connection connection= RabbitMQUtils.getConnection();
            try { 
                Channel channel=connection.createChannel();
                channel.queueDeclare("work",true,false,false,null);
                for(int i=0;i<1000;i++){ 
                    channel.basicPublish("","work",null,("hello"+i).getBytes());
                }
                RabbitMQUtils.closeConnectionAndChanel(channel,connection);
            } catch (IOException e) { 
                e.printStackTrace();
            }
    
        }
    }

### 创建消费者 ###

package com.ak.demo_2;
    
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class Consumer_1 { 
        public static void main(String[] args) { 
            try { 
                // 获取连接对象
                Connection connection=RabbitMQUtils.getConnection();
                // 获取连接中通道
                Channel channel=connection.createChannel();
                channel.queueDeclare("work",true,false,false,null);
                channel.basicConsume("",true, new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println(new String(body));
                    }
                });
               // RabbitMQUtils.closeConnectionAndChanel(channel,connection);
            } catch (IOException e) { 
                e.printStackTrace();
            } finally { 
            }
    
        }
    }

package com.ak.demo_2;
    
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class Consumer_2 { 
        public static void main(String[] args) { 
            try { 
                // 获取连接对象
                Connection connection=RabbitMQUtils.getConnection();
                // 获取连接中通道
                Channel channel=connection.createChannel();
                channel.queueDeclare("work",true,false,false,null);
                channel.basicConsume("",true, new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println(new String(body));
                    }
                });
               // RabbitMQUtils.closeConnectionAndChanel(channel,connection);
            } catch (IOException e) { 
                e.printStackTrace();
            } finally { 
            }
    
        }
    }

> `总结:默认情况下，RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。`

### 消息自动确认机制 ###

> 如何实现能者多劳的任务模型。需要手动确认信息

package com.ak.demo_2;
    
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class Consumer_1_Information { 
        public static void main(String[] args) { 
            try { 
                // 获取连接对象
                Connection connection=RabbitMQUtils.getConnection();
                // 获取连接中通道
                Channel channel=connection.createChannel();
                channel.basicQos(1);
                channel.queueDeclare("workquene",true,false,false,null);
                channel.basicConsume("",false, new DefaultConsumer(channel){ 
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                        System.out.println(new String(body));
                        // 手动确认 ，参数1：手动确认信息标识，参数2：false 每次确认一个
                        channel.basicAck(envelope.getDeliveryTag(),false);
                    }
                });
               // RabbitMQUtils.closeConnectionAndChanel(channel,connection);
            } catch (IOException e) { 
                e.printStackTrace();
            } finally { 
            }
    
        }
    }

## 4.3 fanout 模型 ##

![image-20191126213115873][]

在广播模式下，消息发送流程是这样的：

*  可以有多个消费者
 *  每个**消费者有自己的queue**（队列）
 *  每个**队列都要绑定到Exchange**（交换机）
 *  **生产者发送的消息，只能发送到交换机**，交换机来决定要发给哪个队列，生产者无法决定。
 *  交换机把消息发送给绑定过的所有队列
 *  队列的消费者都能拿到消息。实现一条消息被多个消费者消费

### 开发生产者 ###

package com.ak.fanout;
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    
    public class Provider { 
        public static void main(String[] args) throws IOException { 
            Connection connection= RabbitMQUtils.getConnection();
            Channel channel=connection.createChannel();
            // 参数一：为交换机名称，参数二：fanout 为交换机
            channel.exchangeDeclare("joker","fanout");
            // 发送信息
            channel.basicPublish("joker","",null,"fanout type message".getBytes());
    
            RabbitMQUtils.closeConnectionAndChanel(channel,connection);
    
        }
    }

### 开发消费者 ###

package com.ak.fanout;
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class MyConsumer_3 { 
        public static void main(String[] args) throws IOException { 
            Connection connection = RabbitMQUtils.getConnection();
            Channel channel=connection.createChannel();
            // 通道绑定交换机
            channel.exchangeDeclare("joker","fanout");
            // 临时队列
            String queneName=channel.queueDeclare().getQueue();
            // 绑定交换机和队列
            channel.queueBind(queneName,"joker","");
            // 消费信息
            channel.basicConsume(queneName,true,new DefaultConsumer(channel){ 
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println(new String(body));
                }
            });
        }
    }

## 4.4 Routing ##

### 4.4.1 直连 ###

`在Fanout模式中，一条消息，会被所有订阅的队列都消费。但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。`

在Direct模型下：

*  队列与交换机的绑定，不能是任意绑定了，而是要指定一个`RoutingKey`（路由key）
 *  消息的发送方在 向 Exchange发送消息时，也必须指定消息的 `RoutingKey`。
 *  Exchange不再把消息交给每一个绑定的队列，而是根据消息的`Routing Key`进行判断，只有队列的`Routingkey`与消息的 `Routing key`完全一致，才会接收到消息

流程:

![image-20191126220145375][]

图解：

*  P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
 *  X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
 *  C1：消费者，其所在队列指定了需要routing key 为 error 的消息
 *  C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

> 我希望只有错误日志的时候，才能存储到磁盘
> 
> 其他日志在控制台打印

#### 开发生产者 ####

package com.ak.routedirect;
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    import java.util.Random;
    
    public class Provider { 
        public static void main(String[] args) throws IOException { 
            Connection connection= RabbitMQUtils.getConnection();
            Channel channel=connection.createChannel();
            // 参数一：为交换机名称，参数二：fanout 为交换机
            channel.exchangeDeclare("log_router","direct");
            String []routeKey={ "error","info","waring","debug"};
    
            // 发送信息
            for(int i=0;i<10;i++){ 
                int temp=new Random().nextInt(100)%4;
                channel.basicPublish("log_router",routeKey[temp],null,("发送的信息为 "+routeKey[temp]).getBytes());
            }
            RabbitMQUtils.closeConnectionAndChanel(channel,connection);
    
        }
    }

#### 开发消费者 ####

package com.ak.routedirect;
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class Consumer_1 { 
        public static void main(String[] args) throws IOException { 
            Connection connection= RabbitMQUtils.getConnection();
            // 获取连接中通道
            Channel channel=connection.createChannel();
            channel.exchangeDeclare("log_router","direct");
            // 临时队列
            String queneName=channel.queueDeclare().getQueue();
            // 绑定交换机和队列
            channel.queueBind(queneName,"log_router","error");
            // 消费信息
            channel.basicConsume(queneName,true,new DefaultConsumer(channel){ 
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("消费者2 "+new String(body));
                }
            });
        }
    }

package com.ak.routedirect;
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class Consumer_2 { 
        public static void main(String[] args) throws IOException { 
            Connection connection= RabbitMQUtils.getConnection();
            // 获取连接中通道
            Channel channel=connection.createChannel();
            channel.exchangeDeclare("log_router","direct");
            // 临时队列
            String queneName=channel.queueDeclare().getQueue();
            // 绑定交换机和队列
            channel.queueBind(queneName,"log_router","error");
            channel.queueBind(queneName,"log_router","info");
            channel.queueBind(queneName,"log_router","waring");
            channel.queueBind(queneName,"log_router","debug");
    
            // 消费信息
            channel.basicConsume(queneName,true,new DefaultConsumer(channel){ 
    
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("消费者1 "+new String(body));
                }
            });
        }
    }

#### 检验 ####

![739e43ad7f8ab9f398282f43adc083b2.png][]

### 4.4.2 Routing 之订阅模型-Topic ###

`Topic`类型的`Exchange`与`Direct`相比，都是可以根据`RoutingKey`把消息路由到不同的队列。只不过`Topic`类型`Exchange`可以让队列在绑定`Routing key` 的时候使用通配符！这种模型`Routingkey` 一般都是由一个或多个单词组成，多个单词之间以”.”分割，例如： `item.insert`  
![image-20191127121900255][]

# 统配符
    		* (star) can substitute for exactly one word.    匹配不多不少恰好1个词
    		# (hash) can substitute for zero or more words.  匹配一个或多个词
    # 如:
    		audit.#    匹配audit.irs.corporate或者 audit.irs 等
        audit.*   只能匹配 audit.irs

#####  #####

#### 创建生产者 ####

package com.ak.routeTopic;
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.Connection;
    
    import java.io.IOException;
    import java.util.Random;
    
    public class Provider { 
        public static void main(String[] args) throws IOException { 
            Connection connection= RabbitMQUtils.getConnection();
            Channel channel=connection.createChannel();
            // 参数一：为交换机名称，参数二：fanout 为交换机
            String channelName="log_top";
            channel.exchangeDeclare(channelName,"topic");
            String []routeKey={ "user.save","user.add","admin.add","admin.save"};
    
            for (int i=0;i<10;i++){ 
                int temp=new Random().nextInt(100)%4;
                channel.basicPublish(channelName,routeKey[temp],null,("这个是topics 发布的信息"+routeKey[temp]).getBytes());
            }
            RabbitMQUtils.closeConnectionAndChanel(channel,connection);
        }
    }

#### 创建消费者 ####

package com.ak.routeTopic;
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class Consumer_1 { 
        public static void main(String[] args) throws IOException { 
            Connection connection= RabbitMQUtils.getConnection();
            // 获取连接中通道
            Channel channel=connection.createChannel();
            String channelName="log_top";
            channel.exchangeDeclare(channelName,"topic");
            // 临时队列
            String queneName=channel.queueDeclare().getQueue();
            // 绑定交换机和队列
    
            channel.queueBind(queneName,channelName,"user.*");
            // 消费信息
            channel.basicConsume(queneName,true,new DefaultConsumer(channel){ 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("消费者user 的信息 "+new String(body));
                }
            });
        }
    }

package com.ak.routeTopic;
    
    import com.ak.utils.RabbitMQUtils;
    import com.rabbitmq.client.*;
    
    import java.io.IOException;
    
    public class Consumer_2 { 
        public static void main(String[] args) throws IOException { 
            Connection connection= RabbitMQUtils.getConnection();
            // 获取连接中通道
            Channel channel=connection.createChannel();
            String channelName="log_top";
            channel.exchangeDeclare(channelName,"topic");
            // 临时队列
            String queneName=channel.queueDeclare().getQueue();
            // 绑定交换机和队列
    
            channel.queueBind(queneName,channelName,"admin.*");
            // 消费信息
            channel.basicConsume(queneName,true,new DefaultConsumer(channel){ 
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { 
                    System.out.println("消费者admin 的信息 "+new String(body));
                }
            });
        }
    }

#### 检查 ####

![e7f9ada0f8689b04de9238fb1965093d.png][]

[image-20191126165840602]: /images/20221119/f69831e660fc4ed4ab860deaad39a79c.png
[2cbbf0faa6ffe671193b778b87db7238.png]: /images/20221119/44ae31b3de944e689983fddb683d525c.png
[2703f2f8be2dc38a8ebcffb288a7ff58.png]: /images/20221119/1f3a326b5d1042bb9d7a06393437898e.png
[dacd115c3644d07a25658673816eb2be.png]: /images/20221119/f58bad24324a483c95fbe7df5fed849c.png
[f45ecc984db1fb9f5e8fed23023783a8.png]: /images/20221119/29e9927f263d46d3a5cbd9f14d84cf81.png
[image-20200314221002008]: /images/20221119/07f1a3f4817548e5acab8a2eb0f4e09a.png
[image-20191126213115873]: /images/20221119/6826b2ca491a49ec86e139a13ab0be88.png
[image-20191126220145375]: /images/20221119/a6e0ad9a6be147de903d5c7faede5042.png
[739e43ad7f8ab9f398282f43adc083b2.png]: /images/20221119/d3bae2edf40a4eddbfaae815a1814a36.png
[image-20191127121900255]: /images/20221119/2398301776ef4f3e849a77e695653200.png
[e7f9ada0f8689b04de9238fb1965093d.png]: /images/20221119/8166c36b755f41f3ae2bb0ea6c7be5d8.png