通达信本地数据开放接口_Java笔记使用RabbitMQ的Java接口实现Publish/Subscribe 订阅模式

基本概念

模型如上；

一个生产者，多个消费者；

每个消费者都有自己的队列；

生产者没有直接把消息发送到队列，而是发送到通达信本地数据开放接口,交换机，通过通达信本地数据开放接口,交换机转发到队列；

每个队列都要绑定到通达信本地数据开放接口,交换机上；

生产者发送的消息经过通达信本地数据开放接口,交换机到达队列就能实现一个消息被多个消费者消费；

这里要注意：

在RabbitMQ中通达信本地数据开放接口,交换机没有存储能力，只有队列里面有；

代码与实例

程序运行截如下：

生产者：

两个消费者：

RabbitMQ相关：

关键源码如下：

package ps;

import com.rabbitmq.client.*;
import util.ConnectionUtils;

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

public class Recv1 {

    private static final String QUEUE_NAME = 'test_queue_fanout_sms';
    private static final String EXCHAGE_NAME = 'test_exchange_fanout';
    public static void main(String[] args) throws IOException, TimeoutException {

        Connection connection = ConnectionUtils.getConnect();
        final Channel channel = connection.createChannel();

        //声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        //绑定到通达信本地数据开放接口,交换机
        channel.queueBind(QUEUE_NAME, EXCHAGE_NAME, '');
        channel.basicQos(1);

        Consumer consumer = new DefaultConsumer(channel){

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {

                String msg = new String(body, 'utf-8');
                System.out.println('[1] Recv msg : ' + msg);

                try {

                    Thread.sleep(1000);
                } catch (InterruptedException e) {

                    e.printStackTrace();
                } finally {

                    System.out.println('[1] done');
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

package ps;


import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import util.ConnectionUtils;

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

public class Send {

    private static final String EXCHANGE_NAME = 'test_exchange_fanout';

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

        Connection connection = ConnectionUtils.getConnect();
        Channel channel = connection.createChannel();

        //声明通达信本地数据开放接口,交换机
        channel.exchangeDeclare(EXCHANGE_NAME, 'fanout');   //并分配

        String msg = 'Hello ps';
        channel.basicPublish(EXCHANGE_NAME, '', null, msg.getBytes());

        System.out.println('Send : ' + msg);

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

源码下载地址：

文章为作者独立观点，不代表股票量化交易软件接口观点