JMS系列(三)-java操做JMS Topic实例

时间 2019-12-04

标签 jms 系列 java topic 实例栏目 Java 繁體版

原文原文链接

在上一篇介绍如何经过java往jms消息队列里面写消息和读取消息，本文介绍如何经过java往jms主题里写消息和读取消息。java

消息发布

一样将消息发布到主题中，须要通过如下步骤web

链接jms服务器
获取链接工厂(Connection Factory)
经过链接工厂建立主题链接（TopicConnection)
经过主题链接建立主题会话（TopicSession)
经过主题会话建立主题发布者（Publisher)
建立消息(Message)
经过发布者将消息发送到主题中

代码实现：segmentfault

package asan.demo.jms;

import java.util.Hashtable;
import javax.jms.JMSException;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class JMSTopicSender {
    private TopicPublisher sender = null;
    private TopicSession session = null;
    private static final String JMS_FACTORY_JNDI = "jms/jms_test_connection_factory1";
    private static final String JMS_TOPIC_JNDI = "jms/jms_test_topic";

    public JMSTopicSender() {
        super();
    }

    public void sendMessage(String msg) {
        TextMessage textMsg;
        try {
            if (this.sender == null) {
                this.init();
            }
            textMsg = session.createTextMessage();
            textMsg.setText(msg);
            sender.send(textMsg);
        } catch (JMSException e) {
            e.printStackTrace();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
    //    1. 链接jms服务器
    //    2. 获取链接工厂(Connection Factory)
    //    3. 经过链接工厂建立主题链接（TopicConnection)
    //    4. 经过主题链接建立主题会话（TopicSession)
    //    5. 经过主题会话建立主题发布者（Publisher)
    //    6. 建立消息(Message)
    //    7. 经过发布者将消息发送到主题中
    private void init() throws NamingException, JMSException {
        Hashtable properties = new Hashtable();
        properties.put(Context.INITIAL_CONTEXT_FACTORY,
                       "weblogic.jndi.WLInitialContextFactory");
        properties.put(Context.PROVIDER_URL, "t3://127.0.0.1:7101");
        properties.put(Context.SECURITY_PRINCIPAL, "weblogic");
        properties.put(Context.SECURITY_CREDENTIALS, "weblogic1");
        InitialContext ctx = new InitialContext(properties);
        TopicConnectionFactory jmsFactory =
            (TopicConnectionFactory)ctx.lookup(JMS_FACTORY_JNDI);
        TopicConnection jmsConn = jmsFactory.createTopicConnection();
        session = jmsConn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
        Topic topic = (Topic)ctx.lookup(JMS_TOPIC_JNDI);
        sender = session.createPublisher(topic);
    }
    
    public static void main(String[]cmd){
        JMSTopicSender sender=new JMSTopicSender();
        sender.sendMessage("hello jms topic");
    }
}

与队列不一样的是，此时主题没有订阅者，那么该消息就不存储在主题中，即便后面有订阅者订阅了该主题，也没法接收订阅前的消息。bash

消息订阅

从主题中订阅消息，须要通过如下步骤：服务器

链接jms服务器
获取链接工厂(Connection Factory)
经过链接工厂建立主题链接（TopicConnection)
经过主题链接建立主题会话（TopicSession)
经过主题会话建立订阅者（Subscriber)
接收消息(Message)

代码实现：session

package asan.demo.jms;

import java.util.Hashtable;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.naming.Context;
import javax.naming.InitialContext;
import javax.naming.NamingException;

public class JMSTopicReciver {
    private TopicSubscriber reciver = null;
    private static final String JMS_FACTORY_JNDI = "jms/jms_test_connection_factory1";
    private static final String JMS_TOPIC_JNDI = "jms/jms_test_topic";
    public JMSTopicReciver() {
        super();
    }
    public void reciveMessage() {
        try {
            if (this.reciver == null) {
                this.init();
            }
            System.out.println("waiting to recive message from jms topic "+JMS_TOPIC_JNDI);
            while(true){
                Message msg=reciver.receive();
                if(msg instanceof TextMessage){
                    TextMessage textMsg=(TextMessage)msg;
                    System.out.println("recive jms message:"+textMsg.getText());
                }
            }
        } catch (JMSException e) {
            e.printStackTrace();
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }
    //    1. 链接jms服务器
    //    2. 获取链接工厂(Connection Factory)
    //    3. 经过链接工厂建立主题链接（TopicConnection)
    //    4. 经过主题链接建立主题会话（TopicSession)
    //    5. 经过主题会话建立订阅者（Subscriber)
    //    6. 接收消息(Message)
    private void init() throws NamingException, JMSException {
        Hashtable properties = new Hashtable();
        properties.put(Context.INITIAL_CONTEXT_FACTORY,
                       "weblogic.jndi.WLInitialContextFactory");
        properties.put(Context.PROVIDER_URL, "t3://127.0.0.1:7101");
        properties.put(Context.SECURITY_PRINCIPAL, "weblogic");
        properties.put(Context.SECURITY_CREDENTIALS, "weblogic1");
        InitialContext ctx = new InitialContext(properties);
        TopicConnectionFactory jmsFactory =
            (TopicConnectionFactory)ctx.lookup(JMS_FACTORY_JNDI);
        TopicConnection jmsConn = jmsFactory.createTopicConnection();
        TopicSession session = jmsConn.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
        Topic topic = (Topic)ctx.lookup(JMS_TOPIC_JNDI);
        reciver = session.createSubscriber(topic);
        jmsConn.start();
    }
    
    public static void main(String[]cmd){
        JMSTopicReciver consumer=new JMSTopicReciver();
        consumer.reciveMessage();
    }
}

运行代码，此时运行以前消息发布的代码，能够在控制台看到发送的消息dom

一样，稍微修改下上一篇文章的客户端程序，使整个过程看起来更清晰，修改后的客户端代码以下：this

package asan.demo.jms;

import java.util.Scanner;

public class JMSClient {
    public JMSClient() {
        super();
    }

    public static void help() {
        System.out.println("Usage:java -jar JMSClient.jar sender/reciver/topicSender/topicReciver");
        System.out.println("sender:向jms队列发送消息");
        System.out.println("reciver:从队列中取出消息");
        System.out.println("topicSender:向jms主题发送消息");
        System.out.println("topicReciver:从主题中取出消息");
    }

    public static void main(String[] cmd) {
        if (cmd.length == 0) {
            help();
            return;
        }
        String mode = cmd[0];
        if ("sender".equalsIgnoreCase(mode)) {
            JMSSender sender = new JMSSender();
            Scanner sc = new Scanner(System.in);
            while (true) {
                System.out.println("input you message(input end to exist):");
                String msg = sc.nextLine();
                if ("end".equalsIgnoreCase(msg)) {
                    return;
                }
                sender.sendMessage(msg);
                System.out.println("message send success");
            }
        } else if ("reciver".equalsIgnoreCase(mode)) {
            JMSReciver consumer = new JMSReciver();
            consumer.reciveMessage();
        } else if ("topicSender".equalsIgnoreCase(mode)) {
            JMSTopicSender sender = new JMSTopicSender();
            Scanner sc = new Scanner(System.in);
            while (true) {
                System.out.println("input you message(input end to exist):");
                String msg = sc.nextLine();
                if ("end".equalsIgnoreCase(msg)) {
                    return;
                }
                sender.sendMessage(msg);
                System.out.println("message send success");
            }
        } else if ("topicReciver".equalsIgnoreCase(mode)) {
            JMSTopicReciver consumer = new JMSTopicReciver();
            consumer.reciveMessage();
        }
    }
}

打包运行，执行如下命令将客户端做为主题发布者spa

java -jar JMSDemo.jar topicSender

新建窗口，执行如下命令将客户端做为主题订阅者code

java -jar JMSDemo.jar topicReciver

在发布者上发送消息，在订阅者上看到收到的消息

能够再多开一个窗口，再运行一个订阅者

登陆weblogic控制台，进入domain->Services->Messaging->JMS Modules->jms_test_module->jms_test_topic能够查看当前主题订阅统计信息