RabbitMQ02_大数据系统

文章目录

①. Simple简单模式
②. Publish/Subscribe发布与订阅模式
③. Routing路由模式
④. Topics通配符模式
⑤. Work模式 - 轮询模式(Round-Robin)
⑥. Work模式 - 公平分发(Fair-Dispatch)
⑦. 模式总结

①. Simple简单模式

①. 简单模式分析如下:
生产者:发送消息到RabbitMQ队列(simple_queue)
消费者:接收RabbitMQ队列消息

②. 导入对于的依赖、工具类的抽取

    
         com.rabbitmq
         amqp-client
         5.6.0

public class ConnectionUtil {
    public static Connection getConnection() throws IOException, TimeoutException {
        //1. 创建连接工厂(设置RabbitMQ的连接参数)；
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机；默认localhost
        connectionFactory.setHost("139.198.169.136");
        //连接端口；默认5672
        connectionFactory.setPort(5672);
        //虚拟主机；默认/
        connectionFactory.setVirtualHost("/myvitrualhost");
        //用户名；默认guest
        connectionFactory.setUsername("tang");
        //密码；默认guest
        connectionFactory.setPassword("9602111022yxTZ@");
        //2. 创建连接；
        return connectionFactory.newConnection();
    }
}

③. 编写生产者
注意:设置虚拟机,一个mq服务可以设置多个虚拟机,每个虚拟机就相当于一个独立的mq

public class Producer {
    static final String QUEUE_NAME = "simple_queue";
    public static void main(String[] args) throws Exception{
        //1. 创建连接工厂(设置RabbitMQ的连接参数)；
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //主机；默认localhost
        connectionFactory.setHost("139.198.169.136");
        //连接端口；默认5672
        connectionFactory.setPort(5672);
        //虚拟主机；默认/
        connectionFactory.setVirtualHost("/myvitrualhost");
        //用户名；默认guest
        connectionFactory.setUsername("tang");
        //密码；默认guest
        connectionFactory.setPassword("9602111022yxTZ@");
        //2. 创建连接；
        Connection connection = connectionFactory.newConnection("简单模式生产者");
        //3. 创建频道；
        Channel channel = connection.createChannel();
        //4. 声明队列:如果队列在mq中没有则要创建
        
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        //5. 发送消息；
        String message = "你好！小兔纸。";

        
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println("已发送消息:" + message);
        //6. 关闭资源
        channel.close();
        connection.close();
    }
}

④. 编写消费者
需要持续监听队列消息,所以不要关闭资源

public class Consumer {
    public static void main(String[] args) throws Exception {
        //1. 创建连接工厂；
        //2. 创建连接；(抽取一个获取连接的工具类)
        Connection connection = ConnectionUtil.getConnection();
        //3. 创建频道；
        Channel channel = connection.createChannel();
        //4. 声明队列；
        
        //channel.queueDeclare(Producer1.QUEUE_NAME, true, false, false, null);
        //5. 创建消费者(接收消息并处理消息)；
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为:" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为:" + envelope.getExchange());
                //消息id
                //消息id,mq在channel中用来标识消息的id,可用来确认消息已接收
                System.out.println("消息id为:" + envelope.getDeliveryTag());
                //接收到的消息
                System.out.println("接收到的消息为:" + new String(body, "utf-8"));
            }
        };

        //6. 监听队列
        
        channel.basicConsume(Producer.QUEUE_NAME, true, defaultConsumer);
    }
}

⑤. 总结如下

⑥. 图形化操作简单模式

②. Publish/Subscribe发布与订阅模式

①. 应用场景(用户通知,当用户充值成功或转账完成系统通知用户,通知方式有短信、邮件多种方法)
②. 模式说明
发布与订阅模式:一个消息可以被多个消费者接收。一个消费者对于的队列,该队列只能被一个消费者监听。使用了订阅模式中交换机类型为:广播

③. 生产者

public class Producer {
    //交换机名称
    static final String FANOUT_EXCHANGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";
    //队列名称
    static final String FANOUT_QUEUE_2 = "fanout_queue_2";

    public static void main(String[] args) throws Exception {
        //1. 创建连接；
        Connection connection = ConnectionUtil.getConnection();
        //2. 创建频道；
        Channel channel = connection.createChannel();
        //3. 声明交换机；参数1:交换机名称,参数2:交换机类型(fanout,direct,topic)
        channel.exchangeDeclare(FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT);

        //4. 声明队列；
        
        channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);
        channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);

        //5. 队列绑定到交换机；参数1:队列名称,参数2:交换机名称,参数3:路由key
        channel.queueBind(FANOUT_QUEUE_1, FANOUT_EXCHANGE, "");
        channel.queueBind(FANOUT_QUEUE_2, FANOUT_EXCHANGE, "");

        //6. 发送消息；
        for(int i = 1; i<=10; i++) {
            String message = "你好！小兔纸。发布订阅模式 --- " + i;
            
            channel.basicPublish(FANOUT_EXCHANGE, "", null, message.getBytes());
            System.out.println("已发送消息:" + message);
        }
        //6. 关闭资源
        channel.close();
        connection.close();
    }
}

④. 消费者

public class Consumer {
    private static Runnable runnable=()->{
        try{
            //1. 创建连接；(抽取一个获取连接的工具类)
            Connection connection = ConnectionUtil.getConnection();
            //2. 创建频道；
            Channel channel = connection.createChannel();
            //3.申明队列queue存储消息
            
            // 这里如果queue已经被创建过一次了,可以不需要定义
            // channel.queueDeclare("queue1", false, false, false, null);
            //6. 创建消费者(接收消息并处理消息)；
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //路由key
                    System.out.println("路由key为:" + envelope.getRoutingKey());
                    //交换机
                    System.out.println("交换机为:" + envelope.getExchange());
                    //消息id
                    System.out.println("消息id为:" + envelope.getDeliveryTag());
                    //接收到的消息
                    System.out.println("消费者1 --- 接收到的消息为:" + new String(body, "utf-8"));
                }
            };
            //6. 监听队列
            
            channel.basicConsume(Thread.currentThread().getName(), true, defaultConsumer);
        }catch (Exception e){
            e.getLocalizedMessage();
        }
    };
    public static void main(String[] args) {
        new Thread(runnable, Producer.FANOUT_QUEUE_1).start();
        new Thread(runnable, Producer.FANOUT_QUEUE_2).start();
    }
}

⑤. 测试:

启动所有消费者,然后使用生产者发送消息;在每个消费者对应的控制台可以查看到生产者发送的所有消息;达到广播的效果
在执行完测试代码后,其实到RabbitMQ的管理后台找到Exchanges选项卡,点击fanout_exchange的交换机,可以查看到如下的绑定

③. Routing路由模式

①. Routing模式和Publish/subscibe有啥区别？
Routing模式要求队列在绑定交换机时要指定routingkey,消息会转发到符合routingkey的队列
②. Routing路由模式

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

③. 生产者

public class Producer {
    //交换机名称
    static final String DIRECT_EXCHANGE = "direct_exchange";
    //队列名称
    static final String DIRECT_QUEUE_INSERT = "direct_queue_insert";
    //队列名称
    static final String DIRECT_QUEUE_UPDATE = "direct_queue_update";

    public static void main(String[] args) throws Exception {
        //1. 创建连接；
        Connection connection = ConnectionUtil.getConnection();
        //2. 创建频道；
        Channel channel = connection.createChannel();
        //3. 声明交换机；参数1:交换机名称,参数2:交换机类型(fanout,direct,topic)
        channel.exchangeDeclare(DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT);
        //4. 声明队列；
        
        channel.queueDeclare(DIRECT_QUEUE_INSERT, true, false, false, null);
        channel.queueDeclare(DIRECT_QUEUE_UPDATE, true, false, false, null);

        //5. 队列绑定到交换机；参数1:队列名称,参数2:交换机名称,参数3:路由key
        channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHANGE, "insert");
        channel.queueBind(DIRECT_QUEUE_INSERT, DIRECT_EXCHANGE, "delete");
        channel.queueBind(DIRECT_QUEUE_UPDATE, DIRECT_EXCHANGE, "update");

        //6. 发送消息；
        String message = "你好！小兔纸。路由模式;routing key为insert";

        
        channel.basicPublish(DIRECT_EXCHANGE, "insert", null, message.getBytes());
        System.out.println("已发送消息:" + message);

        message = "你好！小兔纸。路由模式 ；routing key 为 update ";

        
        channel.basicPublish(DIRECT_EXCHANGE, "update", null, message.getBytes());
        message = "你好！小兔纸。路由模式 ；routing key 为 delete ";

        
        channel.basicPublish(DIRECT_EXCHANGE, "delete", null, message.getBytes());
        System.out.println("已发送消息:" + message);
        //6. 关闭资源
        channel.close();
        connection.close();
    }
}

④. 消费者

public class Consumer {
    static Runnable runnable=()->{
        try{
            //1. 创建连接；(抽取一个获取连接的工具类)
            Connection connection = ConnectionUtil.getConnection();
            //2. 创建频道；
            Channel channel = connection.createChannel();
            //3. 创建消费者(接收消息并处理消息)；
            DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    //路由key
                    System.out.println("路由key为:" + envelope.getRoutingKey());
                    //交换机
                    System.out.println("交换机为:" + envelope.getExchange());
                    //消息id
                    System.out.println("消息id为:" + envelope.getDeliveryTag());
                    //接收到的消息
                    System.out.println("消费者1 --- 接收到的消息为:" + new String(body, "utf-8"));
                }
            };
            //4. 监听队列
            
            String name = Thread.currentThread().getName();
            System.out.println("线程的名字:"+name);
            channel.basicConsume(name, true, defaultConsumer);
        }catch (Exception e){
            e.printStackTrace();
        }
    };
    public static void main(String[] args) throws Exception {
        new Thread(runnable,Producer.DIRECT_QUEUE_INSERT).start();
        new Thread(runnable,Producer.DIRECT_QUEUE_UPDATE).start();
    }
}

⑤. 测试
在执行完测试代码后,其实到RabbitMQ的管理后台找到Exchanges选项卡,点击direct_exchange的交换机,可以查看到如下的绑定:

④. Topics通配符模式

①. 通配符规则:

# :匹配一个或多个词
* :匹配不多不少恰好1个词
举例:
item.#:能匹配item.insert.abc或者item.insert
item.*:只能匹配item.insert

②. 生产者

public class Producer {
    //交换机名称
    static final String TOPIC_EXCHAGE = "topic_exchage";
    //队列名称
    static final String TOPIC_QUEUE_1 = "topic_queue_1";
    //队列名称
    static final String TOPIC_QUEUE_2 = "topic_queue_2";

    public static void main(String[] args) throws Exception {
        //1. 创建连接；
        Connection connection = ConnectionUtil.getConnection();
        //2. 创建频道；
        Channel channel = connection.createChannel();
        //3. 声明交换机；参数1:交换机名称,参数2:交换机类型(fanout,direct,topic)
        channel.exchangeDeclare(TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
        channel.queueDeclare(Producer.TOPIC_QUEUE_1, true, false, false, null);

        //5. 队列绑定到交换机上
        channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.update");
        channel.queueBind(Producer.TOPIC_QUEUE_1, Producer.TOPIC_EXCHAGE, "item.delete");

        //5. 队列绑定到交换机上
        channel.queueDeclare(Producer.TOPIC_QUEUE_2, true, false, false, null);
        channel.queueBind(Producer.TOPIC_QUEUE_2, Producer.TOPIC_EXCHAGE, "item.*");
        //6. 发送消息；
        String message = "商品新增。通配符模式 ；routing key 为 item.insert ";

        
        channel.basicPublish(TOPIC_EXCHAGE, "item.insert", null, message.getBytes());
        System.out.println("已发送消息:" + message);

        message = "商品修改。通配符模式 ；routing key 为 item.update ";

        
        channel.basicPublish(TOPIC_EXCHAGE, "item.update", null, message.getBytes());
        System.out.println("已发送消息:" + message);

        message = "商品删除。通配符模式 ；routing key 为 item.delete ";

        
        channel.basicPublish(TOPIC_EXCHAGE, "item.delete", null, message.getBytes());
        System.out.println("已发送消息:" + message);

        //6. 关闭资源
        channel.close();
        connection.close();
    }
}

③. 消费者

public class Consumer1 {
    public static void main(String[] args) throws Exception {
        //1. 创建连接；(抽取一个获取连接的工具类)
        Connection connection = ConnectionUtil.getConnection();
        //2. 创建频道；
        Channel channel = connection.createChannel();
        //3. 声明交换机
        channel.exchangeDeclare(Producer.TOPIC_EXCHAGE, BuiltinExchangeType.TOPIC);
    
        //4. 创建消费者(接收消息并处理消息)；
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为:" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为:" + envelope.getExchange());
                //消息id
                System.out.println("消息id为:" + envelope.getDeliveryTag());
                //接收到的消息
                System.out.println("消费者1 --- 接收到的消息为:" + new String(body, "utf-8"));
            }
        };
        //6. 监听队列
        
        channel.basicConsume(Producer.TOPIC_QUEUE_1, true, defaultConsumer);
    }
}

public class Consumer2 {
    public static void main(String[] args) throws Exception {
        //1. 创建连接；(抽取一个获取连接的工具类)
        Connection connection = ConnectionUtil.getConnection();
        //2. 创建频道；
        Channel channel = connection.createChannel();
        
        //3. 创建消费者(接收消息并处理消息)；
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //路由key
                System.out.println("路由key为:" + envelope.getRoutingKey());
                //交换机
                System.out.println("交换机为:" + envelope.getExchange());
                //消息id
                System.out.println("消息id为:" + envelope.getDeliveryTag());
                //接收到的消息
                System.out.println("消费者1 --- 接收到的消息为:" + new String(body, "utf-8"));
            }
        };
        //6. 监听队列
        
        channel.basicConsume(Producer.TOPIC_QUEUE_2, true, defaultConsumer);
    }
}

④. 测试:
在执行完成测试代码后,其实到RabbitMQ的管理后台找到Exchanges选项卡,点击topix_exchange的交换机,可以查看到如下的绑定:

⑤. Work模式 - 轮询模式(Round-Robin)

①. 特点:该模式接收消息是当有多个消费者接入时,消息的分配模式是一个消费者分配一条,直至消息消费完成(轮询的方式)
②. 生产者代码展示:

public class Producer {
    static final String QUEUE_NAME = "work_queue_round-robin";
    public static void main(String[] args) throws Exception {
        //2. 创建连接；
        Connection connection = ConnectionUtil.getConnection();
        //3. 创建频道；
        Channel channel = connection.createChannel();
        //4. 声明队列；
        
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        for(int  i=0;i<10;i++) {
            //5. 发送消息；
            String message = "你好！小兔纸work-----" + i;
            
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("已发送消息:" + message);
        }
        //6. 关闭资源
        channel.close();
        connection.close();
    }
}

③. 消费者代码

public class Consumer1 {
    public static void main(String[] args) throws Exception {
        //1. 创建连接工厂；
        //2. 创建连接；(抽取一个获取连接的工具类)
        Connection connection = ConnectionUtil.getConnection();
        //3. 创建频道；
        final Channel channel = connection.createChannel();
        System.out.println("consumer1开始消费");
        //5. 创建消费者(接收消息并处理消息)；
        DefaultConsumer defaultConsumer = 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, "utf-8"));
            }
        };
        //6. 监听队列
        
        channel.basicConsume(Producer.QUEUE_NAME, true, defaultConsumer);
    }
}

public class Consumer2 {
    public static void main(String[] args) throws Exception {
        //1. 创建连接工厂；
        //2. 创建连接；(抽取一个获取连接的工具类)
        Connection connection = ConnectionUtil.getConnection();
        //3. 创建频道；
        final Channel channel = connection.createChannel();
        System.out.println("consumer2开始消费");
        //5. 创建消费者(接收消息并处理消息)；
        DefaultConsumer defaultConsumer = 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, "utf-8"));
            }
        };
        //6. 监听队列
        
        channel.basicConsume(Producer.QUEUE_NAME, true, defaultConsumer);
    }
}

⑥. Work模式 - 公平分发(Fair-Dispatch)

①. 公平分发:根据消费者的消费能力进行公平分发,处理快的处理的多,处理慢的处理的少；按劳分配
②. 生产者代码展示

public class Producer {
    static final String QUEUE_NAME = "work_queue_fair";
    public static void main(String[] args) throws Exception {
        //2. 创建连接；
        Connection connection = ConnectionUtil.getConnection();
        //3. 创建频道；
        Channel channel = connection.createChannel();
        //4. 声明队列；
        
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        for(int  i=0;i<10;i++) {
            //5. 发送消息；
            String message = "你好！小兔纸work-----" + i;
            
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("已发送消息:" + message);
        }
        //6. 关闭资源
        channel.close();
        connection.close();
    }
}

③. 消费者代码

public class Consumer1 {
    public static void main(String[] args) throws Exception {
        //1. 创建连接工厂；
        //2. 创建连接；(抽取一个获取连接的工具类)
        Connection connection = ConnectionUtil.getConnection();
        //3. 创建频道；
        final Channel channel = connection.createChannel();
        channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
        System.out.println("fair的方式consumer1开始消费");
        //每次可以预期多少个消息
        channel.basicQos(1);
        //4. 创建消费者(接收消息并处理消息)；
        //fair的方式一定要将应答方式改成手动应答
        DefaultConsumer defaultConsumer = 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, "utf-8"));
                try {
                    Thread.sleep(200);
                    //确认消失
                    
                    channel.basicAck(envelope.getDeliveryTag(),false);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        //6. 监听队列
        
        channel.basicConsume(Producer.QUEUE_NAME, false, defaultConsumer);
    }
}

public class Consumer2 {
    public static void main(String[] args) throws Exception {
        //1. 创建连接工厂；
        //2. 创建连接；(抽取一个获取连接的工具类)
        Connection connection = ConnectionUtil.getConnection();
        //3. 创建频道；
        final Channel channel = connection.createChannel();
        channel.queueDeclare(Producer.QUEUE_NAME, true, false, false, null);
        System.out.println("fair的方式consumer2开始消费");
        //每次可以预期多少个消息
        channel.basicQos(1);
        //5. 创建消费者(接收消息并处理消息)；
        DefaultConsumer defaultConsumer = 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, "utf-8"));
                try {
                    Thread.sleep(5);
                    //确认消失
                    
                    channel.basicAck(envelope.getDeliveryTag(),false);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        //6. 监听队列
        
        channel.basicConsume(Producer.QUEUE_NAME, false, defaultConsumer);
    }
}

⑦. 模式总结

①. 简单模式:一个生产者一个队列,一个消息被一个消费者接受
②. 工作队列模式:一个队列可以有多个消费者；消费者之间是竞争关系
③. 发布与订阅模式:使用了广播的交换机(fanout),可以将一个消息发送到所有有队列对应的消费者
④. 路由模式:使用了定向交换机(Direct)；根据消息路由key与队列路由key进行比较,一致则队列可接收消息
⑤. 通配符模式:使用了通配符(Topic)交换机,根据消息路由key与队列路由key进行匹配(#,*),一致则队列可接收消息
⑥. 模式官网

RabbitMQ02

大数据系统相关栏目本月热门文章