消息中间件_ActiveMQ消息类型

Posted 2022-08-11 51ma

Activemq消息类型
JMS规范中的消息类型包括TextMessage、MapMessage、ObjectMessage、BytesMessage、和StreamMessage等五种。ActiveMQ也有对应的实现，下面我们结合Spring JMS分别来看一下五种消息类型的收发代码。
1、TextMessage

/**
     * 向指定Destination发送text消息
     * @param destination
     * @param message
     */
    public void sendTxtMessage(Destination destination, final String message)
        if(null == destination)
            destination = jmsTemplate.getDefaultDestination();
        
        jmsTemplate.send(destination, new MessageCreator() 
            public Message createMessage(Session session) throws JMSException 
                return session.createTextMessage(message);
            
        );
        System.out.println("springJMS send text message...");
    

View Code

2、MapMessage

/**
     * 向指定Destination发送map消息
     * @param destination
     * @param message
     */
    public void sendMapMessage(Destination destination, final String message)
        if(null == destination)
            destination = jmsTemplate.getDefaultDestination();
        
        jmsTemplate.send(destination, new MessageCreator() 
            public Message createMessage(Session session) throws JMSException 
                MapMessage mapMessage = session.createMapMessage();
                mapMessage.setString("msgId",message);
                return mapMessage;
            
        );
        System.out.println("springJMS send map message...");
    

View Code

3、ObjectMessage

/**
     * 向指定Destination发送序列化的对象
     * @param destination
     * @param object object 必须序列化
     */
    public void sendObjectMessage(Destination destination, final Serializable object)
        if(null == destination)
            destination = jmsTemplate.getDefaultDestination();
        
        jmsTemplate.send(destination, new MessageCreator() 
            public Message createMessage(Session session) throws JMSException 
                return session.createObjectMessage(object);
            
        );
        System.out.println("springJMS send object message...");
    

View Code

4、BytesMessage

/**
     * 向指定Destination发送字节消息
     * @param destination
     * @param bytes
     */
    public void sendBytesMessage(Destination destination, final byte[] bytes)
        if(null == destination)
            destination = jmsTemplate.getDefaultDestination();
        
        jmsTemplate.send(destination, new MessageCreator() 
            public Message createMessage(Session session) throws JMSException 
                BytesMessage bytesMessage = session.createBytesMessage();
                bytesMessage.writeBytes(bytes);
                return bytesMessage;

            
        );
        System.out.println("springJMS send bytes message...");
    

View Code

5、StreamMessage

/**
     * 向默认队列发送Stream消息
     */
    public void sendStreamMessage(Destination destination) 
        jmsTemplate.send(new MessageCreator() 
            public Message createMessage(Session session) throws JMSException 
                StreamMessage message = session.createStreamMessage();
                message.writeString("stream string");
                message.writeInt(11111);
                return message;
            
        );
        System.out.println("springJMS send Strem message...");
    

View Code

消息接收部分处理：

/**
     * 根据消息类型进行对应的处理
     * @param destination 消息发送/接收共同的Destination
     * @throws JMSException
     */
    public void receive(Destination destination) throws JMSException 
        Message message = jmsTemplate.receive(destination);

        // 如果是文本消息
        if (message instanceof TextMessage) 
            TextMessage tm = (TextMessage) message;
            System.out.println("from" + destination.toString() + " get textMessage：\t" + tm.getText());
        

        // 如果是Map消息
        if (message instanceof MapMessage) 
            MapMessage mm = (MapMessage) message;
            System.out.println("from" + destination.toString() + " get textMessage：\t" + mm.getString("msgId"));
        

        // 如果是Object消息
        if (message instanceof ObjectMessage) 
            ObjectMessage om = (ObjectMessage) message;
            ExampleUser exampleUser = (ExampleUser) om.getObject();
            System.out.println("from" + destination.toString() + " get ObjectMessage：\t"
                    + ToStringBuilder.reflectionToString(exampleUser));
        

        // 如果是bytes消息
        if (message instanceof BytesMessage) 
            byte[] b = new byte[1024];
            int len = -1;
            BytesMessage bm = (BytesMessage) message;
            while ((len = bm.readBytes(b)) != -1) 
                System.out.println(new String(b, 0, len));
            
        

        // 如果是Stream消息
        if (message instanceof StreamMessage) 
            StreamMessage sm = (StreamMessage) message;
            System.out.println(sm.readString());
            System.out.println(sm.readInt());
        
    

View Code