JMS的5中消息类型

r囧r小猫 2022-06-14 05:07 104阅读 0赞

前一篇博文简单介绍了[activemq的demo][activemq_demo]示例，里面传递的是文本消息TextMessage，其实jms除了可以传递文本消息，还可以传递流消息StreamMessage，键值对消息MapMessage，对象消息ObjectMessage以及字节BytesMessage。

即JMS共有五中消息类型

TextMessage  主体包含字符串

BytesMessage  主体包含连续字节流

MapMessage  主体包含键值对

StreamMessage主体包含流

ObjectMessage  主体包含对象

话不多说了，还是直接上代码更直观一些。

今天的代码涉及三个类文件，消息生产者JMSProMessType.java， 消息消费者JMSConMessType.java 和 学生实体类Student.java

Student

package com.zhangqi.jms;
    
    import java.io.Serializable;
    
    /**
     * 学生实体类
     * @author zhangqi
     */
    public class Student implements Serializable {
    
    	/**
    	 *  Student类一定要实现serializable接口。
    	 *  一个类只有实现了serializable才是可以序列化的，通俗的讲实现了serializable接口后我
    	 *  们将可以把这个类，在网络上进行发送，或者将这个类存入到硬盘，序列化的目的就是保存一个对象。
    	 */
    	private static final long serialVersionUID = -245693081951194415L;
    	
    	private String id;
    	private String name;
    	
    	public Student (String id, String name) {
    		this.id = id;
    		this.name = name;
    	}
    	
    	public String toString() {
    		return id + ", " + name;
    	}
    }

JMSProMessType

package com.zhangqi.jms;
    
    import javax.jms.BytesMessage;
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MapMessage;
    import javax.jms.MessageProducer;
    import javax.jms.ObjectMessage;
    import javax.jms.Session;
    import javax.jms.StreamMessage;
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * 消息生产者
     * @author zhangqi
     */
    public class JMSProMessType {
    
    	// 默认连接用户名
        private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;
        // 默认连接密码
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;
        // 默认连接地址
        private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;
    
        public static void main(String[] args) {
            // 连接工厂
            ConnectionFactory connectionFactory;
            Connection connection = null;
            Session session;
            Destination destination;
            MessageProducer messageProducer;
            
            // 实例化连接工厂
            connectionFactory = new ActiveMQConnectionFactory(JMSProMessType.USERNAME, JMSProMessType.PASSWORD, JMSProMessType.BROKEURL);
    
            try {
                // 通过连接工厂获取连接
                connection = connectionFactory.createConnection();
                // 启动连接
                connection.start();
                // 创建session
                session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE);
                // 创建一个名称为HelloWorld的消息队列
                destination = session.createQueue("HelloWorld");
                // 创建消息生产者
                messageProducer = session.createProducer(destination);
                
                // 发送消息
                sendStreamMessage(session, messageProducer);
    
                session.commit();
            } catch (Exception e) {
                e.printStackTrace();
            }finally{
                if(connection != null){
                    try {
                        connection.close();
                    } catch (JMSException e) {
                        e.printStackTrace();
                    }
                }
            }
    
        }
    
        public static void sendObjectMessage(Session session,MessageProducer messageProducer) throws Exception{
            
            // 创建一个对象
        	Student stu = new Student("201701", "lilei");
            ObjectMessage message = session.createObjectMessage(stu);
            
            // 通过消息生产者发出消息
            messageProducer.send(message);
            System.out.println("发送消息： " + stu.toString());
        }
        
        public static void sendMapMessage(Session session,MessageProducer messageProducer) throws Exception{
            
        	// 创建map消息对象
            MapMessage message = session.createMapMessage();
            message.setString("name", "zhangqi");
            message.setString("company", "Pingan");
            
            // 通过消息生产者发出消息
            messageProducer.send(message);
        }
        
        public static void sendBytesMessage(Session session,MessageProducer messageProducer) throws Exception{
            
        	// 创建map消息对象
            BytesMessage message = session.createBytesMessage();
            String str = "This is a bytes message";
            message.writeBytes(str.getBytes());
            
            // 通过消息生产者发出消息
            messageProducer.send(message);
        }
        
        public static void sendStreamMessage(Session session,MessageProducer messageProducer) throws Exception{
            
        	// 创建map消息对象
        	StreamMessage message = session.createStreamMessage();
            String str = "This is a stream message";
            message.writeString(str);
            message.writeInt(2017);
            
            // 通过消息生产者发出消息
            messageProducer.send(message);
        }
    }

JMSConMessType

package com.zhangqi.jms;
    
    import javax.jms.BytesMessage;
    import javax.jms.Connection;
    import javax.jms.ConnectionFactory;
    import javax.jms.Destination;
    import javax.jms.JMSException;
    import javax.jms.MapMessage;
    import javax.jms.Message;
    import javax.jms.MessageConsumer;
    import javax.jms.ObjectMessage;
    import javax.jms.Session;
    import javax.jms.StreamMessage;
    
    import org.apache.activemq.ActiveMQConnection;
    import org.apache.activemq.ActiveMQConnectionFactory;
    
    /**
     * 消息消费者
     * @author zhangqi
     */
    public class JMSConMessType {
    
    	private static final String USERNAME = ActiveMQConnection.DEFAULT_USER;//默认连接用户名
        private static final String PASSWORD = ActiveMQConnection.DEFAULT_PASSWORD;//默认连接密码
        private static final String BROKEURL = ActiveMQConnection.DEFAULT_BROKER_URL;//默认连接地址
    
        public static void main(String[] args) {
            ConnectionFactory connectionFactory;//连接工厂
            Connection connection = null;//连接
    
            Session session; // 会话 接受或者发送消息的线程
            Destination destination; // 消息的目的地
    
            MessageConsumer messageConsumer; // 消息的消费者
    
            //实例化连接工厂
            connectionFactory = new ActiveMQConnectionFactory(JMSConMessType.USERNAME, JMSConMessType.PASSWORD, JMSConMessType.BROKEURL);
    
            try {
                // 通过连接工厂获取连接
                connection = connectionFactory.createConnection();
                connection.start(); //启动连接
                // 创建session
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                // 创建一个连接HelloWorld的消息队列
                destination = session.createQueue("HelloWorld");
                // 创建消息消费者
                messageConsumer = session.createConsumer(destination);
                
                getMessage(messageConsumer);
    
            } catch (JMSException e) {
                e.printStackTrace();
            }
    
        }
        
        /**
         * 接收消息
         * @param messageConsumer
         * @throws JMSException
         */
        public static void getMessage(MessageConsumer messageConsumer) throws JMSException {
        	Message message = messageConsumer.receive();
            
            if (message instanceof ObjectMessage) {  
            	ObjectMessage objMessage = (ObjectMessage) message;
            	Student stu = (Student) objMessage.getObject();
                System.out.println("接收信息： " + stu.toString()); 
            } 
            
            if (message instanceof MapMessage) {
            	MapMessage mapMessage = (MapMessage) message;
            	String name = mapMessage.getString("name");
            	String company = mapMessage.getString("company");
            	System.out.println("name = " + name);
            	System.out.println("company = " + company);
            }
            
            if (message instanceof BytesMessage) {
            	BytesMessage bytesMessage = (BytesMessage) message;
            	byte[] b = new byte[1024];
                int len = -1;
                while ((len = bytesMessage.readBytes(b)) != -1) {
                  System.out.println(new String(b, 0, len));
                }
            }
            
            if (message instanceof StreamMessage) {
            	StreamMessage streamMessage = (StreamMessage) message;
            	String str = streamMessage.readString();
            	int num = streamMessage.readInt();
            	System.out.println(str + ", " + num);
            }
        }
    }

[activemq_demo]: http://blog.csdn.net/magi1201/article/details/72700109