JMS实现简单聊天程序

使用了JMS的开源实现activemq消息中间件。各版本都行

分为如下几步完成。

 

  第一步：安装activemq

  (1)下载解压到E:\open_source\activeMQ\apache-activemq-5.5.0，并在环境变量配置ACTIVEMQ_HOME

  (2)因为apache-activemq-5.5.0需要使用slf4j-1.5.11版本的jar包,下载slf4j

 

  第二步：编写相应的代码，如下：

  package ch02.chat;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Properties;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
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.jms.TopicSubscriber;
import javax.naming.Context;
import javax.naming.InitialContext;


public class Chat implements MessageListener {

    private TopicSession pubSession;
    private TopicPublisher publisher;
    private TopicConnection connection;
    private String username;

    // 用于初始化chat的构造函数
    public Chat(String topicFactory, String topicName, String username)
            throws Exception {
//      Properties env=new Properties();
//      env.put(Context.SECURITY_PRINCIPAL, "system");
//      env.put(Context.SECURITY_CREDENTIALS, "manager");
//      env.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.activemq.jndi.ActiveMQInitialContextFactory");
//      env.put(Context.PROVIDER_URL, "tcp://localhost:61616");

        // 使用jndi.properties文件获得一个JNDI连接
        InitialContext ctx = new InitialContext();

        // 查找一个jms连接工厂并创建连接
        TopicConnectionFactory conFactory = (TopicConnectionFactory) ctx
                .lookup(topicFactory);
        TopicConnection connection = conFactory.createTopicConnection();

        // 创建两个JMS会话对象
        TopicSession pubSession = connection.createTopicSession(false,
                Session.AUTO_ACKNOWLEDGE);
        TopicSession subSession = connection.createTopicSession(false,
                Session.AUTO_ACKNOWLEDGE);

        // 查找一个JMS主题
        Topic chatTopic = (Topic) ctx.lookup(topicName);

        // 创建一个JMS发布者和订阅者,createSubscriber中附加的参数是一个消息
        // 选择器(null)和noLocal标记的一个真值,它表明这个发布者生产的消息不应被他自己所消费
        TopicPublisher publisher = pubSession.createPublisher(chatTopic);
        TopicSubscriber subscriber = subSession.createSubscriber(chatTopic,
                null, true);

        // 设置一个JMS消息侦听器
        subscriber.setMessageListener(this);

        // 初始化Chat应用程序变量
        this.connection = connection;
        this.pubSession = pubSession;
        this.publisher = publisher;
        this.username = username;

        // 启动JMS连接，允许传递消息
        connection.start();
    }

    // 接受来自TopicSubscriber的消息
    public void onMessage(Message message) {
        try {
            TextMessage textMessage = (TextMessage) message;
            System.out.println(textMessage.getText());

        } catch (JMSException jmse) {
            jmse.printStackTrace();
        }
    }

    // 使用发布者创建并发送消息
    protected void writeMessage(String text) throws JMSException {
        TextMessage message = pubSession.createTextMessage();
        message.setText(username + " : " + text);
        publisher.publish(message);
    }

    // 关闭JMS连接
    public void close() throws JMSException {
        connection.close();
    }

    // 运行聊天客户端
    public static void main(String[] args) {
        try {
            if (args.length != 3) {
                System.out.println("Factory.Topic.or username missing");
            }
            String topicFactory="TopicCF";
            String topicName="topic1";
            String username="lyb";
            Chat chat = new Chat(topicFactory, topicName, username);

            // 从命令行读取
            BufferedReader commandLine = new BufferedReader(
                    new InputStreamReader(System.in));

            // 循环
            while (true) {
                String s = commandLine.readLine();
                if (s.equalsIgnoreCase("exit")) {
                    chat.close();
                    System.exit(0);
                } else {
                    chat.writeMessage(s);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        ;
    }

}

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Properties;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
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.jms.TopicSubscriber;
import javax.naming.Context;
import javax.naming.InitialContext;


public class Chat implements MessageListener {

	private TopicSession pubSession;
	private TopicPublisher publisher;
	private TopicConnection connection;
	private String username;

	// 用于初始化chat的构造函数
	public Chat(String topicFactory, String topicName, String username)
			throws Exception {
//		Properties env=new Properties();
//		env.put(Context.SECURITY_PRINCIPAL, "system");
//		env.put(Context.SECURITY_CREDENTIALS, "manager");
//		env.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.activemq.jndi.ActiveMQInitialContextFactory");
//		env.put(Context.PROVIDER_URL, "tcp://localhost:61616");
		
		// 使用jndi.properties文件获得一个JNDI连接
		InitialContext ctx = new InitialContext();

		// 查找一个jms连接工厂并创建连接
		TopicConnectionFactory conFactory = (TopicConnectionFactory) ctx
				.lookup(topicFactory);
		TopicConnection connection = conFactory.createTopicConnection();

		// 创建两个JMS会话对象
		TopicSession pubSession = connection.createTopicSession(false,
				Session.AUTO_ACKNOWLEDGE);
		TopicSession subSession = connection.createTopicSession(false,
				Session.AUTO_ACKNOWLEDGE);

		// 查找一个JMS主题
		Topic chatTopic = (Topic) ctx.lookup(topicName);

		// 创建一个JMS发布者和订阅者,createSubscriber中附加的参数是一个消息
		// 选择器(null)和noLocal标记的一个真值,它表明这个发布者生产的消息不应被他自己所消费
		TopicPublisher publisher = pubSession.createPublisher(chatTopic);
		TopicSubscriber subscriber = subSession.createSubscriber(chatTopic,
				null, true);

		// 设置一个JMS消息侦听器
		subscriber.setMessageListener(this);

		// 初始化Chat应用程序变量
		this.connection = connection;
		this.pubSession = pubSession;
		this.publisher = publisher;
		this.username = username;

		// 启动JMS连接，允许传递消息
		connection.start();
	}

	// 接受来自TopicSubscriber的消息
	public void onMessage(Message message) {
		try {
			TextMessage textMessage = (TextMessage) message;
			System.out.println(textMessage.getText());

		} catch (JMSException jmse) {
			jmse.printStackTrace();
		}
	}

	// 使用发布者创建并发送消息
	protected void writeMessage(String text) throws JMSException {
		TextMessage message = pubSession.createTextMessage();
		message.setText(username + " : " + text);
		publisher.publish(message);
	}

	// 关闭JMS连接
	public void close() throws JMSException {
		connection.close();
	}

	// 运行聊天客户端
	public static void main(String[] args) {
		try {
			if (args.length != 3) {
				System.out.println("Factory.Topic.or username missing");
			}
			String topicFactory="TopicCF";
			String topicName="topic1";
			String username="lyb";
			Chat chat = new Chat(topicFactory, topicName, username);

			// 从命令行读取
			BufferedReader commandLine = new BufferedReader(
					new InputStreamReader(System.in));

			// 循环
			while (true) {
				String s = commandLine.readLine();
				if (s.equalsIgnoreCase("exit")) {
					chat.close();
					System.exit(0);
				} else {
					chat.writeMessage(s);
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
		;
	}

}

     将以上的类复制一份，放在同一个包下面。一份代码作为消息的发布方，一份代码作为消息的订阅方。

     整个程序的工程结构组织见下图

 

    

    

 

  第三步、配置jndi文件

 

 

  java.naming.factory.initial=org.apache.activemq.jndi.ActiveMQInitialContextFactory

  java.naming.provider.url=tcp://localhost\:61616

  java.naming.security.principal=system

  java.naming.security.credentials=manager

  connectionFactoryNames=TopicCF

  topic.topic1=jms.topic1

 

 第四步、启动activeMQ，分别运行Chat.java和CopyChat.java,可以在两个程序之间交互信息