[大数据]RabbitMQ 探究（一）

工作中用的RabbitMQ ,但并没有深入探究它里面的所有消息队列类型，今天有时间来整体的尝试一番！

第一种消息模型: 一个生产者、一个消费者，不使用交换机！

首先需要引入包：

     <!--rabbitmq 依赖客户端-->
        <dependency>
            <groupId>com.rabbitmq</groupId>
            <artifactId>amqp-client</artifactId>
            <version>5.8.0</version>
        </dependency>

创建工具类：

package 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 RabbitMqUtil {

    //创建连接MQ的工厂对象
    private static ConnectionFactory connectionFactory;

    //静态代码块
    static{
        connectionFactory=new ConnectionFactory();
        connectionFactory.setHost("192.168.56.10");
        connectionFactory.setPort(5672);
        //设置连接哪个虚拟主机
        connectionFactory.setVirtualHost("/ems");
        //设置访问虚拟主机的用户名和密码
        connectionFactory.setUsername("ems");
        connectionFactory.setPassword("123");
    }

    //启动连接
    public static Connection getConnection() throws IOException, TimeoutException {
        return connectionFactory.newConnection();
    }

    //关闭连接
    public static void closeConnectionAndChanel(Channel channel,Connection connection) throws IOException, TimeoutException {
        if(channel==null){
            channel.close();
            connection.close();
        }
    }


}

生产者：

package rabbitMq;

import Utils.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

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

//生产者
public class TestRabbitMqP {
    public static void main(String[] args) throws IOException, TimeoutException {

        //传递的消息
        String mes="HELLO WORD !!";

        //通过工具获取连接对象
        Connection connection = RabbitMqUtil.getConnection();
        //获取连接中通道
        Channel channel = connection.createChannel();
        /**
         * 绑定对应的消息队列
         * 参数：
         * 1、队列名称 如果队列不存在，则自动创建
         * 2、是否需要持久化
         * 3、是否独占队列
         * 4、是都消息完成后删除队列
         * 5、额外附加参数
         */
        channel.queueDeclare("hello",false,false,false,null);

        /**
         * 消息发布
         * 参数：
         * 1、交换机名称
         * 2、队列名称
         * 3、传递消息的额外设置
         * 4、消息内容
         */
        channel.basicPublish("","hello",null,mes.getBytes());

        //关闭连接
        RabbitMqUtil.closeConnectionAndChanel(channel,connection);
}

}

消费者：

package rabbitMq;

import Utils.RabbitMqUtil;
import com.rabbitmq.client.*;
import com.sun.scenario.effect.impl.sw.sse.SSEBlend_SRC_OUTPeer;

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

//消费者
public class TestRabbitMqC {

    public static void main(String[] args) throws IOException, TimeoutException {

        //传递的消息
        String mes="HELLO WORD !!";

        //通过工具获取连接对象
        Connection connection = RabbitMqUtil.getConnection();
        //获取连接中通道
        Channel channel = connection.createChannel();
        /**
         * 绑定对应的消息队列
         * 参数：
         * 1、队列名称 如果队列不存在，则自动创建
         * 2、是否需要持久化
         * 3、是否独占队列
         * 4、是都消息完成后删除队列
         * 5、额外附加参数
         */
        channel.queueDeclare("hello",false,false,false,null);

        /**
         * 消息订阅
         * 参数：
         * 1、需要消费的队列名称
         * 2、是否开启消息自动队列
         * 3、消费的回调接口
         */
        channel.basicConsume("hello",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));
            }
        });

    }

}