标签：false String 队列交换机 RabbitMQ02 public channel

1.rabbitmq五种消息模型

1.1work消息模型-工作队列模型

工作队列，又称任务队列。主要思想就是避免执行资源密集型任务时，必须等待它执行完成。相反我们稍后完成任务，我们将任务封装为消息并将其发送到队列。在后台运行的工作进程将获取任务并最终执行作业。当你运行许多消费者时，任务将在他们之间共享，但是一个消息只能被一个消费者获取。

工作队列消息模型中的消费者，它们之间是竞争关系。

1.1.1编写生产者

public class WorkQueueProvider {
    public static void main(String[] args) throws IOException,TimeoutException {
        //1.建立连接
        Connection connection = ConnectionUtil.getConnection();
        //2.构建通道
        Channel channel = connection.createChannel();
        //3.构建队列
        channel.queueDeclare("work_queue",false,false,false,null);
        //4.发送消息
        for (int i=0; i<10;i++){
            String msg = "消息："+i;
            channel.basicPublish("","work_queue",null,msg.getBytes());
        }
        //5.关闭连接
        channel.close();
        connection.close();
    }
}

1.1.2编写消费者

public class WorkQueueConsumer {
	public static void main(String[] args) throws IOException,TimeoutException {
        //1.获取连接
        Connection connection = ConnectionUtil.getConnection();
        //2.构建通道
        Channel channel = connection.createChannel();
        //3.接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException{
                //接收处理消息
                String msg = new String(body);
                System.out.println("消费者2："+msg);
                }
            };
            channel.basicConsume("work_queue",true,consumer);
	}
}

1.1.3配置消费者多实例启动

1.1.4.进行测试

两个消费者一同启动，然后发送10条消息
两个消费者各自消费了5条消息，而且各不相同，这就实现了任务的分发

1.1.5.出现的问题

如果消费者1比消费者2的效率要低，一次任务的耗时较长，然而两人最终消费的消息数量是一样的
消费者2大量时间处于空闲状态，消费者1一直忙碌

1.1.6.解决问题，让能者多劳

//工作队列消费者
public class WorkQueueConsumer {
	public static void main(String[] args) throws IOException,TimeoutException {
        //1.获取连接
        Connection connection = ConnectionUtil.getConnection();
        //2.构建通道
        Channel channel = connection.createChannel();
        //设置消费者一次只能处理一条消息，即处理完一条消息才能考虑下一条消息处理，不能提前规划，手动确认才会生效
        channel.basicQos(1);
        //3.接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException{
            //接收处理消息
            try {
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String msg = new String(body);
            System.out.println("消费者1："+msg);
            channel.basicAck(envelope.getDeliveryTag(),false);
        	}
        };
        channel.basicConsume("work_queue",false,consumer);
    }
}

注意：basicQos需要手动确认才会生效

1.2订阅模型分类

在之前的模式中，我们创建了一个工作队列。工作队列背后的假设是：每个任务只被传递给一个工作人员。在这一部分，我们将做一些完全不同的事情 - 我们将会传递一个信息给多个消费者。这种模式被称为“发布/订阅”。

1.2.1订阅模型的特点

1、1个生产者，多个消费者

2、每一个消费者都有自己的一个队列

3、生产者没有将消息直接发送到队列，而是发送到了交换机

4、每个队列都要绑定到交换机

5、生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的

1.2.2交换机的作用和类型

交换机一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

Fanout：广播，将消息交给所有绑定到交换机的队列（发布订阅模式）
Direct：定向，把消息交给符合指定routing key 的队列（路由规则模式）
Topic：通配符，把消息交给符合routing pattern（路由模式）的队列（通配符路由规则模式）

1.2.3订阅模型-Fanout

1.广播模式消息发送的流程:

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

2.定义生产者

public class FanoutProvider {
	public static void main(String[] args) throws IOException,TimeoutException {
        //1.获取连接
        Connection connection = ConnectionUtil.getConnection();
        //2.构建通道
        Channel channel = connection.createChannel();
        //3.构建交换机
        //参数1：交换机名称 参数2：交换机类型 参数3：是否持久化交换机
        channel.exchangeDeclare("fanout_exchange",
        BuiltinExchangeType.FANOUT,false);
        //4.构建队列
        channel.queueDeclare("fanout_queue1",false,false,false,null);
        channel.queueDeclare("fanout_queue2",false,false,false,null);
        //5.交换机和队列绑定
        //参数1：队列名称，参数2：交换机名称 参数3：绑定规则，路由规则，如果是fanout发布订阅模式，是""
        channel.queueBind("fanout_queue1","fanout_exchange","");
        channel.queueBind("fanout_queue2","fanout_exchange","");
        //6.发送消息
        //参数1：交换机名称 参数2：路由规则 参数3：配置参数 参数4：消息
        String msg = "Hello Fanout";
        channel.basicPublish("fanout_exchange","",null,msg.getBytes());
        //7.关闭连接
        channel.close();
        connection.close();
	}
}

3.定义消费者

//发布/订阅模式消费者
public class FanouConsumer {
public static void main(String[] args) throws IOException,TimeoutException {
        //1.获取连接
        Connection connection = ConnectionUtil.getConnection();
        //2.构建通道
        Channel channel = connection.createChannel();
        //3.发送消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException{
                String msg = new String(body);
                System.out.println("消费者2："+msg);
            }
        };
        channel.basicConsume("fanout_queue2",true,consumer);
    }
}

4.进行测试

1.2.4订阅模型-Direct（Routing）路由规则模型

aa.error

bb.error

*.error

在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的 Exchange。在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个 RoutingKey（路由key）

消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。

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

1.定义生产者

public class DirectProvider {
	public static void main(String[] args) throws IOException,TimeoutException {
        //1.获取连接
        Connection connection = ConnectionUtil.getConnection();
        //2.构建通道
        Channel channel = connection.createChannel();
        //3.构建交换机
        //参数1：交换机名称 参数2：交换机类型 参数3：是否持久化交换机
        channel.exchangeDeclare("direct_exchange",
BuiltinExchangeType.DIRECT,false);
        //4.构建队列
        channel.queueDeclare("direct_queue1",false,false,false,null);
        channel.queueDeclare("direct_queue2",false,false,false,null);
        //5.交换机和队列绑定
        //参数1：队列名称，参数2：交换机名称 参数3：绑定规则，路由规则，如果是fanout发布订阅模式，是""
        channel.queueBind("direct_queue1","direct_exchange","error");
        channel.queueBind("direct_queue2","direct_exchange","info");
        channel.queueBind("direct_queue2","direct_exchange","error");
        channel.queueBind("direct_queue2","direct_exchange","warning");
        //6.发送消息
        //参数1：交换机名称 参数2：发送消息的路由规则 参数3：配置参数参数4：消息
        String msg = "Hello Direct";
        channel.basicPublish("direct_exchange","warning",null,msg.getBytes());
        //7.关闭连接
        channel.close();
        connection.close();
    }
}

2.定义消费者

public class DirectConsumer {
    public static void main(String[] args) throws IOException,TimeoutException {
        //1.获取连接
        Connection connection = ConnectionUtil.getConnection();
        //2.构建通道
        Channel channel = connection.createChannel();
        //3.发送消息
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
            {
                String msg = new String(body);
                System.out.println("消费者2："+msg);
            }
        };
        channel.basicConsume("direct_queue2",true,consumer);
    }
}

3.进行测试

会发现发送消息时，如果发送不同的Routing，那就会有不同的消费者接收。

1.2.5订阅模型-Topic通配符路由规则模型

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

1.通配符规则

#：匹配一个或多个词
*：匹配不多不少恰好1个词
audit.#：能够匹配audit.irs.corporate 或者 audit.irs
audit.*：只能匹配audit.irs

2.定义生产者

public class TopicsProvider {
    public static void main(String[] args) throws IOException,TimeoutException {
        //1.获取连接
        Connection connection = ConnectionUtil.getConnection();
        //2.构建通道
        Channel channel = connection.createChannel();
        //3.构建交换机
        //参数1：交换机名称 参数2：交换机类型 参数3：是否持久化交换机
        channel.exchangeDeclare("topics_exchange",
BuiltinExchangeType.TOPIC,false);
        //4.构建队列
        channel.queueDeclare("topics_queue1",false,false,false,null);
        channel.queueDeclare("topics_queue2",false,false,false,null);
        //5.交换机和队列绑定
        //参数1：队列名称，参数2：交换机名称 参数3：绑定规则，路由规则，如果是fanout发布订阅模式，是""
        channel.queueBind("topics_queue1","topics_exchange","super.#");
        channel.queueBind("topics_queue2","topics_exchange","*.info");
        //6.发送消息
        //参数1：交换机名称 参数2：发送消息的路由规则 参数3：配置参数 参数4：消息
        String msg = "Hello Topic";
       		channel.basicPublish("topics_exchange","super.info",null,msg.getBytes());
        //7.关闭连接
        channel.close();
        connection.close();
    }
}

3.定义消费者

public class TopicsConsumer {
	public static void main(String[] args) throws IOException,TimeoutException {
            //1.获取连接
            Connection connection = ConnectionUtil.getConnection();
            //2.构建通道
            Channel channel = connection.createChannel();
            //3.发送消息
            DefaultConsumer consumer = new DefaultConsumer(channel){
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException
                {
                    String msg = new String(body);
                    System.out.println("消费者2："+msg);
                }
            };
        channel.basicConsume("topics_queue2",true,consumer);
    }
}

2.rabbitmq的持久化

消费者的ACK机制。可以防止消费者丢失消息。
但是，如果在消费者消费之前，MQ就宕机了，消息就没了。

要将消息持久化，前提是：队列、Exchange都持久化

2.1交换机持久化

2.2队列持久化

2.3消息持久化

3.springAMQP

Spring-amqp是对AMQP协议的抽象实现，而spring-rabbit 是对协议的具体实现，也是目前的唯一实现。底层使用的就是RabbitMQ。

3.1引入相关依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

3.2编写配置类

@Configuration
public class RabbitMQConfiguration {
    //构建交换机 topic
    @Bean
    public TopicExchange topicExchange(){
    	return ExchangeBuilder.topicExchange("springboot_topic_exchange").durable(true).bu
ild();
    }
    //构建队列
    @Bean
    public Queue topicQueue(){
        //Queue queue = new Queue("springboot_topic_queue");
        return QueueBuilder.durable("springboot_topic_queue").build();
    }
    //交换机和队列绑定
    @Bean
    public Binding topicBinding(@Qualifier("topicQueue") Queue queue,
                                @Qualifier("topicExchange") TopicExchange exchange){
    	return BindingBuilder.bind(queue).to(exchange).with("#.error");
    }
}

3.3编写yml配置文件

spring:
    rabbitmq:
        host: 192.168.200.129
        port: 5672
        username: zhangsan
        password: 123456
        virtual-host: /zhangsan

3.4编写Controller类进行测试

@RestController
public class RabbitMQController {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @RequestMapping("/send")
    public String send(String message){
    //参数1：交换机名称 参数2：路由规则 参数3：发送的消息
    rabbitTemplate.convertAndSend("springboot_topic_exchange","aa.bb.error",me
ssage.getBytes());
    	return "success";
    }
}

3.5编写监听器

@Component
public class RabbitMQListener {
    @RabbitListener(queues = {"Topic_Queue"})
    public void listenerMessage(Message message, Channel channel){ //参数：用来存放从rabbitmq的消息队列中接收的消息的
        String msg = new String(message.getBody());
        System.out.println("接收的消息："+msg);
    }
}

3.6手动ACK操作

application.yml中配置

spring:
    rabbitmq:
        host: 192.168.200.129
        port: 5672
        username: zhangsan
        password: 123456
        virtual-host: /zhangsan
        listener:
        	simple:
        		acknowledge-mode: manual # 手动ack

在消费者中设置

@Component
public class RabbitMQListener {
    @RabbitListener(queues = {"springboot_topic_queue"})
    public void listenerMessage(Message message, Channel channel) throws
IOException {
        byte[] body = message.getBody();
        String msg = new String(body);
        System.out.println("消费者："+msg);
        if ("MLGB".equals(msg)){
        //拒绝确认
        //第三个参数：requeue：重回队列。如果设置为true，则消息重新回到queue，broker会重新发送该消息给消费端
        channel.basicNack(message.getMessageProperties().getDeliveryTag(),false,true);
        }else{
        //手动确认
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
        }
    }
}

标签：false,String,队列,交换机,RabbitMQ02,public,channel
From： https://www.cnblogs.com/jiabaolatiao/p/17512345.html

RabbitMQ02

1.rabbitmq五种消息模型

1.1work消息模型-工作队列模型

1.1.1编写生产者

1.1.2编写消费者

1.1.3配置消费者多实例启动

1.1.4.进行测试

1.1.5.出现的问题

1.1.6.解决问题，让能者多劳

1.2订阅模型分类

1.2.1订阅模型的特点

1.2.2交换机的作用和类型

1.2.3订阅模型-Fanout

1.广播模式消息发送的流程:

2.定义生产者

3.定义消费者

4.进行测试

1.2.4订阅模型-Direct（Routing）路由规则模型

1.定义生产者

2.定义消费者

3.进行测试

1.2.5订阅模型-Topic通配符路由规则模型

1.通配符规则

2.定义生产者

3.定义消费者

2.rabbitmq的持久化

2.1交换机持久化

2.2队列持久化

2.3消息持久化

3.springAMQP

3.1引入相关依赖

3.2编写配置类

3.3编写yml配置文件

3.4编写Controller类进行测试

3.5编写监听器

3.6手动ACK操作

相关文章

赞助商

阅读排行