JMS-TOPIC生产者端(JMS服务器采用activeMQ)

1.首先启动JMS(ACTIVEMQ)服务器

2.编写代码如下:

package com.active;
import java.util.Arrays;
import java.util.Date;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.util.IndentPrinter;

public class ProducerTool {

    private Destination destination;
    private int messageCount = 10;
    private long sleepTime;
    private boolean verbose = true;
    private int messageSize = 255;
    private long timeToLive;
    private String user = ActiveMQConnection.DEFAULT_USER;
    private String password = ActiveMQConnection.DEFAULT_PASSWORD;
    private String url = ActiveMQConnection.DEFAULT_BROKER_URL;
    private String subject = "topic1";
    private boolean topic =true;
    private boolean transacted;
    private boolean persistent =true;//表示是否要持久化消息。

    /*
 ProducerTool [url] broker的地址,默认的是tcp://localhost:61616
    [true|flase] 是否使用topic,默认是false
    [subject] subject的名字,默认是TOOL.DEFAULT
    [durabl] 是否持久化消息,默认是false
    [messagecount] 发送消息数量,默认是10
    [messagesize] 消息长度,默认是255
    [clientID] durable为true的时候,需要配置clientID
    [timeToLive] 消息存活时间
    [sleepTime] 发送消息中间的休眠时间
    [transacte]  是否采用事务

    ConsumerTool [url] broker的地址,默认的是tcp://localhost:61616
    [true|flase] 是否使用topic,默认是false
    [subject] subject的名字,默认是TOOL.DEFAULT
    [durabl] 是否持久化消息,默认是false
    [maxiumMessages] 接受最大消息数量,0表示不限制
    [clientID] durable为true的时候,需要配置clientID
    [transacte]  是否采用事务
    [sleepTime]  接受消息中间的休眠时间,默认是0,onMeesage方法不休眠
    [receiveTimeOut] 接受超时

 */

    public static void main(String[] args) {
        ProducerTool producerTool = new ProducerTool();
        String[] unknown = CommandLineSupport.setOptions(producerTool, args);
        if (unknown.length > 0) {
            System.out.println("Unknown options: " + Arrays.toString(unknown));
            System.exit(-1);
        }
        producerTool.run();
    }

    public void run() {
        Connection connection = null;
        try {
            System.out.println("Connecting to URL: " + url);
            System.out.println("Publishing a Message with size " + messageSize + " to " + (topic ? "topic" : "queue") + ": " + subject);
            System.out.println("Using " + (persistent ? "persistent" : "non-persistent") + " messages");
            System.out.println("Sleeping between publish " + sleepTime + " ms");
            if (timeToLive != 0) {
                System.out.println("Messages time to live " + timeToLive + " ms");
            }

            // Create the connection.首先创建一个连接
            ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(user, password, url);
            connection = connectionFactory.createConnection();
            connection.start();

            // Create the session.创建一个SESSION
            Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);

           //如果是TOPIC的方式则创建TOPIC连接.否则创建QUEUE连接
            if (topic) {
                destination = session.createTopic(subject);
            } else {
                destination = session.createQueue(subject);
            }

            // Create the producer.
            MessageProducer producer = session.createProducer(destination);
            if (persistent) {
                producer.setDeliveryMode(DeliveryMode.PERSISTENT);

            //我们如果设置Mode的方式为DeliveryMode.PERSISTENT.(我采用的是MYSQL的持久化方式)那么服务器所发送的消息将都保存到MYSQL数据库中的activemq_msgs表中
            } else {
                producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

            //如果不是采用持久化保存的方式,那么当服务器重新启动的时候,由于客户端不在线时服务器所发的消息将不再保留.因为那些消息是保存在了服务器内存中的
            }
            if (timeToLive != 0) {
                producer.setTimeToLive(timeToLive);
            }

            // Start sending messages
            sendLoop(session, producer);

            System.out.println("Done.");

            // Use the ActiveMQConnection interface to dump the connection
            // stats.
           // ActiveMQConnection c = (ActiveMQConnection)connection;
           // c.getConnectionStats().dump(new IndentPrinter());

        } catch (Exception e) {
            System.out.println("Caught: " + e);
            e.printStackTrace();
        } finally {
            try {
                connection.close();

            注意发送消息完后要释放资源。
            } catch (Throwable ignore) {
            }
        }
    }

    protected void sendLoop(Session session, MessageProducer producer) throws Exception {

            TextMessage message = session.createTextMessage();
            message.setText("i love wuweiling");
            producer.send(message);
    }
    public void setPersistent(boolean durable) {
        this.persistent = durable;
    }

    public void setMessageCount(int messageCount) {
        this.messageCount = messageCount;
    }

    public void setMessageSize(int messageSize) {
        this.messageSize = messageSize;
    }

    public void setPassword(String pwd) {
        this.password = pwd;
    }

    public void setSleepTime(long sleepTime) {
        this.sleepTime = sleepTime;
    }

    public void setSubject(String subject) {
        this.subject = subject;
    }

    public void setTimeToLive(long timeToLive) {
        this.timeToLive = timeToLive;
    }

    public void setTopic(boolean topic) {
        this.topic = topic;
    }

    public void setQueue(boolean queue) {
        this.topic = !queue;
    }

    public void setTransacted(boolean transacted) {
        this.transacted = transacted;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public void setUser(String user) {
        this.user = user;
    }

    public void setVerbose(boolean verbose) {
        this.verbose = verbose;
    }
}