rabbitmq-5-案例1-简单的案例

2019/09/26 23:35
阅读数 12

最简单的案例:

https://gitee.com/n_zhe/rabbitmq-demo

 

通过简单的例子分析mq是怎样发送和拉取消息的:

 quickStart中的简单案例
    通过简单的案例来分析rabbitmq是怎样发送和拉取消息的
1、查看Procuder.java文件
package com.xsxy.rabbitmq.demo.quickStart;

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

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

/**
 * Procuder
 */
public class Procuder {

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

        // 2、通过工厂创建connection
        Connection connection = connectionFactory.newConnection();

        // 3、通过connection创建一个Channel
        Channel channel = connection.createChannel();

        String exchange = "";
        String routingKey = "test001";
        BasicProperties props = null;
        String msg = "hello rabbit-mq";
        // 4、通过channel发送数据 发送5次数据
        for (int i = 0; i < 5; i++) {
            channel.basicPublish(exchange, routingKey, props, msg.getBytes());
        }

        System.out.println("已经发送消息了");
        // 5、记得要关闭相关的链接
        channel.close();
        connection.close();
    }

}

 

 
    从最开始的ConnectionFactory开始
        ①首先是常见了一个工厂
            进入ConnectionFactory中,可以明显的看到一些基本的参数,例如默认的用户,密码,vhost。。。
            该对象只有一个默认的无参构造,因此具体参数的初始化都在connectionFactory.newConnection()方法中
        ②进入newConnection()方法
public Connection newConnection() throws IOException, TimeoutException {
     return newConnection(this.sharedExecutor, Collections.singletonList(new Address(getHost(), getPort())));
}

  最终调用:

  参数executor默认为null,addrs为mq的地址+端口,clientProvideName为null

  

  继续观察:

FrameHandlerFactory fhFactory = createFrameHandlerFactory();

  其实这个FrameHandlerFactory就是对SocketFactory进行了一次封装:

  

  

  最终返回的为SocketFrameHandler类,该类是对Socket进行了封装:

  

  继续向下走,会使用之前的返回的FrameHander对象new一个AMQConnection对象:

  

 

  然后会调用AMQConnection的start()方法:

  下边这个for循环,第一次启动后就直接return??

  

 

 

  ③返回AMQConnection对象

    通过Connectin来创建Channel,返回ChannelN对象

  ④通过ChannelN对象的basicPublish方法发布消息

    先通过exchange,routingKey等参数构建一个Publish,进而构建一个AMQCommand对象

  

 

   最终:

  

 

 流程:

 

 

  消费端:

package com.xsxy.rabbitmq.demo.quickStart;

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

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;
import com.rabbitmq.client.QueueingConsumer.Delivery;

/**
 * Consumer
 */
public class Consumer {

    public static void main(String[] args) throws IOException, TimeoutException, ShutdownSignalException,
            ConsumerCancelledException, InterruptedException {
        // 1、创建链接工厂ConnectionFactory
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("****");
        connectionFactory.setPort(5672);
        connectionFactory.setVirtualHost("/");
        connectionFactory.setUsername("***");
        connectionFactory.setPassword("***");

        // 1.1 以下两个配置代表是否自动重连) 防止因网络故障导致mq断开
        connectionFactory.setAutomaticRecoveryEnabled(true);
        connectionFactory.setNetworkRecoveryInterval(3000);

        // 2、通过工厂创建connection
        Connection connection = connectionFactory.newConnection();

        // 3、通过connection创建一个Channel
        Channel channel = connection.createChannel();

        // 4、声明一个队列
        // queue 队列的名称 durable 是否持久化 exclusive 是否独占 autoDelete 是否自动删除 arguments
        // 其他的一些参数设置
        String queueName = "test001";
        channel.queueDeclare(queueName, true, false, false, null);

        // 5、常见一个消费者
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

        // 6、 设置channel
        channel.basicConsume(queueName, true, queueingConsumer);

        System.out.println("等待获取消息======");
        // 7、获取消息
        while (true) {
            Delivery delivery = queueingConsumer.nextDelivery();
            String body = new String(delivery.getBody());
            System.out.println("消费端" + body);
        }
    }
}

 

  开始也是获取ConnectionFactory、Connection、Channel,然后通过Channel来操作(不管是生产端还是消费端,mq都是通过channel来进行操作的)

  ①消费端会声明一个消费队列

// 5、常见一个消费者
        QueueingConsumer queueingConsumer = new QueueingConsumer(channel);

  

 

  ②通过Channel将消费这和消息队列关联

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

 

  运行结果:

  消费端:

等待获取消息======

  生产端:

已经发送消息了

当生产端发送消息之后,消费端输出:

等待获取消息======
消费端hello rabbit-mq
消费端hello rabbit-mq
消费端hello rabbit-mq
消费端hello rabbit-mq
消费端hello rabbit-mq

 

 以上代码没有指定exchange,因此rabbitmq server会自动通过默认的exchange(即default exchange)取转发消息,如果生产者的routingkey和消费端的队列名称相同的话,则能够转发成功,否则失败

 

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