`

activemq-topic

阅读更多
http://wangxinchun.iteye.com/blog/2146172
http://wangxinchun.iteye.com/blog/2146120
http://wangxinchun.iteye.com/blog/2145998
http://wangxinchun.iteye.com/blog/2145958
http://wangxinchun.iteye.com/blog/2147335

发布/订阅模式:针对一个发布者,多个订阅者都能接受到消息。

非持久订阅:只有当客户端处于激活状态,也就是和JMS Provider 保持连接状态才能收到发送到某个主题的消息,而当客户端处于离线状态,这个时间段发到主题的消息将会丢失,永远不会收到。
Publisher
/**
 *  消息发布*/
public class Publisher {
    private static final String BROKER_URL = "tcp://localhost:61616";
    private static final Boolean NON_TRANSACTED = false;
    private static final int NUM_MESSAGES_TO_SEND = 100;
    private static final long DELAY = 100;

    public static void main(String[] args) {
        String url = BROKER_URL;
        if (args.length > 0) {
            url = args[0].trim();
        }
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "password", url);
        Connection connection = null;

        try {

            connection = connectionFactory.createConnection();
            connection.start();

            Session session = connection.createSession(NON_TRANSACTED, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createTopic("test-topic");
            MessageProducer producer = session.createProducer(destination);

            for (int i = 0; i < NUM_MESSAGES_TO_SEND; i++) {
                TextMessage message = session.createTextMessage("Message #" + i);
                System.out.println("Sending message #" + i);
                producer.send(message);
                Thread.sleep(DELAY);
            }

            // tell the subscribers we're done
            producer.send(session.createTextMessage("END"));

            producer.close();
            session.close();

        } catch (Exception e) {
            System.out.println("Caught exception!");
        }
        finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    System.out.println("Could not close an open connection...");
                }
            }
        }
    }
}


Subscriber
/**
 * 消息订阅*/
public class Subscriber implements MessageListener {
    private static final String BROKER_URL = "tcp://localhost:61616";
    private static final Boolean NON_TRANSACTED = false;

    private final CountDownLatch countDownLatch;
    public Subscriber(CountDownLatch latch) {
        countDownLatch = latch;
    }

    public static void main(String[] args) {
        String url = BROKER_URL;
        if (args.length > 0) {
            url = args[0].trim();
        }
        System.out.println("\nWaiting to receive messages... Either waiting for END message or press Ctrl+C to exit");
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("admin", "password", url);
        Connection connection = null;
        final CountDownLatch latch = new CountDownLatch(1);

        try {

            connection = connectionFactory.createConnection();
            connection.start();

            Session session = connection.createSession(NON_TRANSACTED, Session.AUTO_ACKNOWLEDGE);
            Destination destination = session.createTopic("test-topic");

            MessageConsumer consumer = session.createConsumer(destination);
            consumer.setMessageListener(new Subscriber(latch));

            latch.await();
            consumer.close();
            session.close();

        } catch (Exception e) {
            System.out.println("Caught exception!");
        }
        finally {
            if (connection != null) {
                try {
                    connection.close();
                } catch (JMSException e) {
                    System.out.println("Could not close an open connection...");
                }
            }
        }
    }

    @Override
    public void onMessage(Message message) {
        try {
            if (message instanceof TextMessage) {
                String text = ((TextMessage) message).getText();
                if ("END".equalsIgnoreCase(text)) {
                    System.out.println("Received END message!");
                    countDownLatch.countDown();
                }
                else {
                    System.out.println("Received message:" +text);
                }
            }
        } catch (JMSException e) {
            System.out.println("Got a JMS Exception!");
        }
    }
}


测试:
1、启动两个 Subscriber 然后启动一个 Publisher
   结果:两个Subscriber  都收到了消息。

2、启动Publisher ,然后再启动 Subscriber
   结果:Publisher发的消息,在 Publisher启动结束和 Subscriber启动结束之间的发送,全部丢失
分享到:
评论

相关推荐

    ActiveMQ-Topic订阅发布模式Demo

    4. **创建Topic**:在ActiveMQ的管理控制台或通过编程方式创建Topic,定义主题名以便生产和消费。 5. **发布消息**:使用MessageProducer对象创建并发送消息到Topic。消息可以是文本、二进制数据或其他复杂类型,...

    boot-example-activemq-topic-2.0.5

    SpringBoot+ActiveMQ-发布订阅 Topic 主题 * 消息消费者(订阅方式)消费该消息 * 消费生产者将发布到topic中,同时有多个消息消费者(订阅)消费该消息 * 这种方式和点对点方式不同,发布到topic的消息会被所有...

    apache-activemq-5.9.0-bin

    1. **消息队列**:ActiveMQ支持多种消息模式,如点对点(Queue)和发布/订阅(Topic)。消息队列确保消息的可靠传输,即使在发送方和接收方之间发生故障时也能保持数据的完整性。 2. **JMS兼容性**:ActiveMQ完全...

    activemq-web-console-5.11.2

    1.一个是admin,用来显示和管理所有的queue、topic、connection等等。 2.一个是demo,有一些使用jms和activemq的简单例子。 3.还有一个fileserver,用来支持通过activemq发送文件时的中转服务器。blob message时配置...

    apache-activemq-5.15.6

    ActiveMQ作为JMS实现,支持点对点(Queue)和发布/订阅(Topic)两种模式。 2. **消息模型**:ActiveMQ支持多种消息模型,包括持久化消息、非持久化消息、事务消息等。持久化消息即使在服务器重启后也能保持,而非...

    apache-activemq-5.16.6-bin.zip

    2. **主题(Topic)与队列(Queue)**: 主题用于发布/订阅模式,多个订阅者可以接收同一条消息;队列则遵循先进先出(FIFO)原则,每个消息仅被一个消费者消费。 3. **持久化(Persistence)**: ActiveMQ支持消息...

    activemq-cpp-library-3.6.0-src.tar.gz_C# ActiveMQ_activemq_activ

    4. **消息模式**:ActiveMQ支持多种消息模式,如点对点(Queue)、发布/订阅(Topic)、请求/响应(Request/Reply)等。这些模式适应了不同场景下的通信需求,如可靠的单次传递、广播或者分布式计算。 5. **高级...

    apache-activemq-5.12.0-bin

    4. **主题与队列**:ActiveMQ支持两种消息模式:主题(Topic)和队列(Queue)。主题用于广播式通信,一条消息可以被多个消费者接收;队列则遵循先入先出(FIFO)原则,消息只能被一个消费者接收。 5. **管理界面**...

    apache-activemq-5.9.0 下载

    3. **JMS支持**:ActiveMQ完全符合JMS 1.1规范,支持Message Queue、Topic、Temporary Queue和Temporary Topic四种类型。用户可以通过JMS API创建连接、会话、生产者、消费者并发送/接收消息。 4. **协议支持**:...

    apache-activemq-5.15.11-bin.tar.gz

    - **队列与主题**:在ActiveMQ中,有两种主要的消息传递模型——队列(Queue)和主题(Topic)。队列采用点对点模型,每个消息只被一个消费者接收;主题采用发布/订阅模型,可以有多个订阅者接收同一消息。 2. **...

    apache-activemq-5.15.0-bin.tar.7z

    MQ是消息中间件,是一种在分布式系统中应用程序借以传递消息的媒介,常用的有ActiveMQ,RabbitMQ,kafka。ActiveMQ是Apache下的开源项目,完全支持JMS1.1和J2EE1.4规范的JMS Provider实现。...2、一对多(topic)

    apache-activemq-5.15.7-bin

    2. **主题(Topic)与队列(Queue)**:ActiveMQ支持两种消息模型——发布/订阅(Publish/Subscribe)和点对点(Point-to-Point)。主题适用于广播式通信,多个订阅者可以接收到相同的消息;队列则遵循FIFO(先进先...

    apache-activemq-5.3.1-bin.tar.gz

    ActiveMQ作为JMS实现,支持点对点(Queue)和发布/订阅(Topic)两种模式。 **3. 安装过程** 下载“apache-activemq-5.3.1-bin.tar.gz”后,你需要解压到一个合适的目录。这可以通过命令行工具如`tar`完成: ```...

    activemq-client-5.8.0.jar

    Apache ActiveMQ是业界广泛使用的开源消息中间件,它遵循开放消息中间件协议(Open Message Middleware,OMM)标准,如JMS(Java Message Service)。在本案例中,我们关注的是`activemq-client-5.8.0.jar`,这是...

    apache-activemq-5.9.1-bin.tar.gz

    此外,ActiveMQ支持多种消息模型,包括点对点(Queue)和发布/订阅(Topic)。点对点模式下,消息只被一个消费者接收;而在发布/订阅模式下,消息可以被多个订阅者消费。这些模型适用于不同的应用场景。 消息传输的...

    apache-activemq-5.15.1-bin.tar.gz

    ActiveMQ实现了JMS 1.1规范,提供了多种消息模型,包括点对点(Queue)和发布/订阅(Topic)。 3. **多种协议支持**:ActiveMQ不仅支持JMS,还支持STOMP、AMQP、XMPP、OpenWire等多种消息传输协议,使其能与其他非...

    apache-activemq-5.15.10-bin.tar.gz

    1. **JMS兼容性** - ActiveMQ符合JMS 1.1规范,支持点对点(Queue)和发布/订阅(Topic)两种消息模型,以及事务处理和持久化消息。 2. **多协议支持** - 支持多种协议,包括OpenWire、AMQP、STOMP、MQTT、WS-...

    apache-activemq-5.15.5-bin.tar.gz

    此外,理解JMS规范中的概念,如队列(Queue)、主题(Topic)、生产者(Producer)、消费者(Consumer)以及消息模型(点对点、发布/订阅),对于有效利用ActiveMQ至关重要。 总之,Apache ActiveMQ是企业级消息...

    Apache.NMS.ActiveMQ-1.7.0-bin

    通过Topic和Queue,可以实现发布/订阅和点对点的消息模式。 总的来说,这个压缩包为C#开发者提供了一套完整的工具,以便他们在.NET环境中无缝地利用Apache ActiveMQ的强大功能,进行高效、灵活的消息通信。对于需要...

    apache-activemq-5.3.0-bin.zip

    - **队列与主题**:ActiveMQ支持两种消息模式——点对点(Queue)和发布/订阅(Topic)。队列保证消息的顺序传递,而主题则允许广播式的消息分发。 2. **ActiveMQ功能**: - **高可用性**:通过集群和复制策略,...

Global site tag (gtag.js) - Google Analytics