package com.hoo.mq.topic;

import javax.jms.DeliveryMode;

import javax.jms.MapMessage;

import javax.jms.Session;

import javax.jms.Topic;

import javax.jms.TopicConnection;

import javax.jms.TopicConnectionFactory;

import javax.jms.TopicPublisher;

import javax.jms.TopicSession;

import org.apache.activemq.ActiveMQConnection;

import org.apache.activemq.ActiveMQConnectionFactory;

/**

 * <b>function:</b> Queue 方式訊息傳送者

 * @author hoojo

 * @createDate 2013-6-19 下午04:34:36

 * @file QueueSender.java

 * @package com.hoo.mq.topic

 * @project ActiveMQ-5.8

 * @blog http://blog.csdn.net/IBM_hoojo

 * @email [email protected]
 

 * @version 1.0

 */

public class TopicSender {

    // 傳送次數

    public static final int SEND_NUM = 5;

    // tcp 地址

    public static final String BROKER_URL = "tcp://localhost:61616";

    // 目標，在ActiveMQ管理員控制檯建立 http://localhost:8161/admin/queues.jsp

    public static final String DESTINATION = "hoo.mq.topic"
 
;

    /**

     * <b>function:</b> 傳送訊息

     * @author hoojo

     * @createDate 2013-6-19 下午12:05:42

     * @param session 會話

     * @param publisher 釋出者

     * @throws Exception

     */    

    public static void sendMessage(TopicSession session, TopicPublisher publisher) throws Exception {

        for (int i = 0; i < SEND_NUM; i++) {

            String message = "傳送訊息第" + (i + 1) + "條";

            MapMessage map = session.createMapMessage();

            map.setString("text", message);

            map.setLong("time", System.currentTimeMillis());

            System.out.println(map);

            publisher.send(map);

        }

    }

    public static void run() throws Exception {

        TopicConnection connection = null;

        TopicSession session = null;

        try {

            // 建立連結工廠

            TopicConnectionFactory factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER, ActiveMQConnection.DEFAULT_PASSWORD, BROKER_URL);

            // 通過工廠建立一個連線

            connection = factory.createTopicConnection();

            // 啟動連線

            connection.start();

            // 建立一個session會話

            session = connection.createTopicSession(Boolean.TRUE, Session.AUTO_ACKNOWLEDGE);

            // 建立一個訊息佇列

            Topic topic = session.createTopic(DESTINATION);

            // 建立訊息傳送者

            TopicPublisher publisher = session.createPublisher(topic);

            // 設定持久化模式

            publisher.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            sendMessage(session, publisher);

            // 提交會話

            session.commit();

        } catch (Exception e) {

            throw e;

        } finally {

            // 關閉釋放資源

            if (session != null) {

                session.close();

            }

            if (connection != null) {

                connection.close();

            }

        }

    }

    public static void main(String[] args) throws Exception {

        TopicSender.run();

    }

}