rabbitmq-5-案例2-简单的案例+exchange

2019/09/29 14:37
阅读数 59

Exchange交换机:

  sendMessage端,发送消息到Exchage1、2, 然后交换机通过路由键,将消息转发给队列queue中,最后客户端从队列中获取消息

 

交换属性:

  name:名称

  type:类型direct、topic、fanout、headers

  durability:是否持久化

  autoDelete:当最后一个绑定到exchange上的队列删除了,自动删除exchage(即当交换机没有和任何队列关联时,将自动删除交换机)

  internal:默认false(当前交换机是否为rabbitmq内部使用)

  arguments:扩展参数

 

direct exchage:直连交换机

  所有发送到Direct Exchage的消息被转发到RouterKey中指定的queue中

 

代码例子:

该例子和之前的quickStart类似,只是指令了交换机和路由键,其他代码相同

ConnectionFactory、Connection、Channel

ConnectionFactory、Connection、Channel,这三个都是RabbitMQ对外提供的API中最基本的对象。不管是服务器端还是客户端都会首先创建这三类对象。
     ConnectionFactory为Connection的制造工厂。

Connection是与RabbitMQ服务器的socket链接,它封装了socket协议及身份验证相关部分逻辑。

Channel是我们与RabbitMQ打交道的最重要的一个接口,大部分的业务操作是在Channel这个接口中完成的,包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

Queue(队列)是RabbitMQ的内部对象,用于存储消息

 

生产端;

package com.xsxy.rabbitmq.demo.direct;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * ProduceDirect
 */
public class ProduceDirect {

    public static void main(String[] args) throws Exception {
        // 1、创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("59.110.232.8");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("woaini");
        connectionFactory.setPassword("woaini");

        // 2、通过工厂获取连接
        Connection connection = connectionFactory.newConnection();

        // 3、获取通道
        Channel channel = connection.createChannel();

        String exchangeName = "directExchange";
        String exchagenType = "direct";
        String queueName = "test.direct.queue";
        String routingKey = "test.direct";
        // 4、声明交换机
        channel.exchangeDeclare(exchangeName, exchagenType, true, false, null);
        // 5、声明队列
        channel.queueDeclare(queueName, true, false, false, null);
        // 6、简历绑定关系
        channel.queueBind(queueName, exchangeName, routingKey);

        // 7发行消息
        for (int i = 0; i < 5; i++) {

            channel.basicPublish(exchangeName, routingKey, null, "testdirectexchage message".getBytes());
        }

        System.err.println("消息已发送===========");

        channel.close();
        connection.close();
    }
}

消费端:

package com.xsxy.rabbitmq.demo.direct;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class ConsumerDirect {

    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1、创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("59.110.232.8");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("woaini");
        connectionFactory.setPassword("woaini");

        // 2、通过Factory创建连接
        Connection connection = connectionFactory.newConnection();

        // 3、创建通道
        Channel channel = connection.createChannel();

        String queueName = "test.direct.queue";
        // 声明队列
        channel.queueDeclare(queueName, true, false, false, null);

        // 4、创建消费去列
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

        // 5、消费消息(非自动确认)
        channel.basicConsume(queueName, false, queueingConsumer);

        // 6、监听消息
        System.out.println("正在监听消息===");
        while (true) {
            QueueingConsumer.Delivery delivery = queueingConsumer.nextDelivery();
            byte[] body = delivery.getBody();
            System.out.println(new String(body));
            // 手动确认
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }

    }
}

运行结果:

 

 

 

 

队列消费者,用于监听队列中的消息。调用nextDelivery方法时,内部实现就是调用队列的take方法。该方法的作用:获取并移除此队列的头部,在元素变得可用之前一直等待(如果有必要)。说白了就是如果没有消息,就处于阻塞状态。

运行结果如下:(生产者、消费者谁先运行都可以)

 

  

相关博客:

https://www.kancloud.cn/longxuan/rabbitmq-arron/117514

 

展开阅读全文
打赏
0
0 收藏
分享
加载中
更多评论
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部