Java 实现RabbitMQ的订阅和发布

 一、关于RabbitMQ搭建和基本概念这里不做介绍，下面给出实用的参考博客

RabbitMQ基础概念及详细介绍参考文档：http://blog.csdn.net/whycold/article/details/41119807

RabbitMQ入门及环境的搭建：http://m.blog.csdn.net/article/details?id=50487028

RabbitMQ网页控制台开启方式：http://blog.csdn.net/spyiu/article/details/24697221

二、RabbitMQ的发布

（1）发布端的连接方法

发布端的连接只需要创建一个ConnectionFactory然后创建一个连接，然后创建一个频道，声明一个路由器，指定名称、模式、及其是否durable

public int BaseConnection(){        /**
         * 创建连接连接到RabbitMQ
         */
        ConnectionFactory factory = new ConnectionFactory();        //设置MabbitMQ所在主机ip或者主机名
        factory.setHost(IP);
        factory.setPort(5672);
        factory.setUsername(user);
        factory.setPassword(password);        //创建一个连接
        try {
            connection = factory.newConnection();
        } catch (IOException e) {
            System.out.println("[x] 请确认输入的IP地址、用户名、密码是否准确！");            return  -1;
        } catch (Exception e) {
            System.out.println("[x] 连接RabbitMQ超时，请重试！");            return  -1;
        }        //创建一个频道
        try {
            channel = connection.createChannel();
        } catch (IOException e) {
            System.out.println("[x] 创建频道出错，请重试！");            return -1;
        }        //声明一个路由器，指定名称、模式、及其是否durable
        try {
            channel.exchangeDeclare(EXCHANGE_NAME,EXCHANGE_TYPE,EXCHANGE_DURABLE);
        } catch (IOException e) {
            System.out.println("[x] 路由器声明失败，请重试！");            return -1;
        }

        System.out.println("[x] 发布消息者成功连接至RabbitMQ !");        return 0;
    }

（2）发布端发布消息的方法

发布端发布消息往指定的exchange（路由）发送的时候需要指定一个routingKey，topic类型时当接收端的bindingKey和routingKey相匹配的时候才能接收到订阅的消息

/**
     * 发布消息
     * @param routingKey routingKey
     * @param msg 消息
     * @return
     */
    public boolean Publish(String routingKey,byte[] msg){            try {
                channel.basicPublish(EXCHANGE_NAME,routingKey,null,msg);
                System.out.println("[x] Sender Sent " + routingKey + " : " + msg);                return true;
            } catch (IOException e) {
                System.out.println("[y] Sender failed to basic publish.");                return false;
            }
    }

三、RabbitMQ的订阅

（1）订阅端的连接方法

订阅端的连接在发布端连接的基础上还需要给该exchange绑定

/**
     * 创建连接连接到RabbitMQ
     * @return   0 means success
     *           -1 means failure
     */
    public int BaseConnection(){
        ConnectionFactory factory = new ConnectionFactory();        //设置MabbitMQ所在主机ip或者主机名
        factory.setHost(IP);
        factory.setPort(5672);
        factory.setUsername(user);
        factory.setPassword(password);        //创建一个连接
        try {
            connection = factory.newConnection();
        } catch (IOException e) {
            System.out.println("[y] 请确认输入的IP地址、用户名、密码是否准确！");            return  -1;
        } catch (Exception e) {
            System.out.println("[y] 连接RabbitMQ超时，请重试！");            return  -1;
        }        //创建一个频道
        try {
            channel = connection.createChannel();
        } catch (IOException e) {
            System.out.println("[y] 创建频道出错，请重试！");            return -1;
        }        //声明一个路由器，指定名称、模式、及其是否durable
        try {
            channel.exchangeDeclare(EXCHANGE_NAME,EXCHANGE_TYPE,EXCHANGE_DURABLE);
        } catch (IOException e) {
            System.out.println("[y] 路由器声明失败，请重试！");            return -1;
        }        //指定一个队列,随机队列名,non-durable,exclusive,not auto-delete//        QUEUE_NAME = MqConfig.QUEUE_NAME;

        Map<String,Object> args = new HashMap();
        args.put("x-message-ttl",message_ttl);        try {
            channel.queueDeclare(QUEUE_NAME,QUEUE_DURABLE,QUEUE_EXCLUSIVE,QUEUE_AUTO_DELETE,args);
        } catch (IOException e) {
            System.out.println("[y] 队列声明失败，请重试！");
        }        
        try {
            channel.queueBind(QUEUE_NAME,EXCHANGE_NAME,bindingKey);
        } catch (IOException e) {
            System.out.println("[y] 队列绑定路由关键字 " + bindingKey + " 时出错!");
        }

        System.out.println("[y] 订阅消息者成功连接至RabbitMQ !");        return 0;
    }

（2）订阅端的订阅方法

/** * 订阅消息函数 */ public boolean Consume(){ System.out.println("[y] Receiver " + QUEUE_NAME + " Waiting for messages.To exit press CTRL+C."); //消息订阅 Consumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { team.iOceanPlus.PB.Config.PBConfig rec=team.iOceanPlus.PB.Config.PBConfig.parseFrom(body); // rec.getConfigTargetDistribution(); // rec.getConfigTargetDistribution().getConfigType(); System.out.println(rec.getConfigTargetDistribution(0)); } }; //消息反馈 try { channel.basicConsume(QUEUE_NAME,true,consumer); return true; } catch (IOException e) { System.out.println("[y] Receiver "+ QUEUE_NAME + " failed to basic consume." ); return false; } }