ActiveMQ发送List类型的消息

朱雀 2020-10-18 07:36 913阅读 0赞

一、生产者代码（主要是把list序列化）

package springboot.activeMQ;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import javax.jms.*;
    import java.io.ByteArrayOutputStream;
    import java.io.IOException;
    import java.io.ObjectOutputStream;
    import java.util.ArrayList;
    import java.util.List;
    
    public class Producer {
    	private static final String url="tcp://127.0.0.1:61616";//服务地址，端口默认61616
    	private static final String queueName="queue-test";//要创建的消息名称
    	private static Logger log = LoggerFactory.getLogger(Producer.class);
    	public static void main(String[] args) throws JMSException {
    		//1.创建ConnectiongFactory,绑定地址
    		ConnectionFactory factory=new ActiveMQConnectionFactory(url);
    		//2.创建Connection
    		Connection connection= factory.createConnection();
    		//3.启动连接
    		connection.start();
    		//4.创建会话
    		Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		//5.创建一个目标
    		Destination destination=session.createQueue(queueName);
    		//6.创建一个生产者
    		MessageProducer producer=session.createProducer(destination);
    		List<String> list = new ArrayList<>();
    		list.add("123");
    		list.add("123");
    		list.add("123");
    		list.add("123");
    		list.add("123");
    		byte[] b = serialize(list);
    		BytesMessage bb = session.createBytesMessage();
    		bb.writeBytes(b);
    		producer.send(bb);
    		connection.close();
    	}
    
    	/**
    	 * 列表序列化（用于Redis整存整取）
    	 * @param value
    	 * @return
    	 */
    	public static <T> byte[] serialize(List<T> value) {
    		if (value == null) {
    			throw new NullPointerException("Can't serialize null");
    		}
    		byte[] rv=null;
    		ByteArrayOutputStream bos = null;
    		ObjectOutputStream os = null;
    		try {
    			bos = new ByteArrayOutputStream();
    			os = new ObjectOutputStream(bos);
    			for(T obj : value){
    				os.writeObject(obj);
    			}
    			os.writeObject(null);
    			os.close();
    			bos.close();
    			rv = bos.toByteArray();
    		} catch (IOException e) {
    			throw new IllegalArgumentException("Non-serializable object", e);
    		}
    		return rv;
    	}
    
    }

二、消费者代码（由于没找到直接获取byte数组的方法，debug里发现content属性里的data便是之前发的数据，所以利用反射获取）

package springboot.activeMQ;
    
    import java.io.ByteArrayInputStream;
    import java.io.IOException;
    import java.io.ObjectInputStream;
    import java.lang.reflect.Field;
    import java.util.ArrayList;
    import java.util.List;
    
    import javax.jms.*;
    
    import org.apache.activemq.ActiveMQConnectionFactory;
    import org.apache.activemq.util.ByteSequence;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    public class Consumer{
    	private static final String url="tcp://127.0.0.1:61616";//端口默认
    	private static final String queueName="queue-test";//要消费的消息名称
    	private static Logger log = LoggerFactory.getLogger(cs.class);
    	public static void main(String[] args) throws JMSException {
    		//1.创建ConnectiongFactory,绑定地址
    		ConnectionFactory factory=new ActiveMQConnectionFactory(url);
    		//2.创建Connection
    		Connection connection= factory.createConnection();
    		//3.启动连接
    		connection.start();
    		//4.创建会话
    		Session session=connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    		//5.创建一个目标
    		Destination destination=session.createQueue(queueName);
    		//6.创建一个消费者
    		MessageConsumer consumer=session.createConsumer(destination);
    		//7.创建一个监听器
    		consumer.setMessageListener(new MessageListener() {
    			public void onMessage(Message arg0) {
    				BytesMessage bm = (BytesMessage)arg0;
    				Field field = null;
    				Class<?> clazz = bm.getClass();
    				for (; clazz != Object.class; clazz = clazz.getSuperclass()) {
    					try {
    						field = clazz.getDeclaredField("content");
    					} catch (Exception e) {
    						// 这里甚么都不能抛出去。
    						// 如果这里的异常打印或者往外抛，则就不会进入
    					}
    				}
    				field.setAccessible(true);
    				byte[] by = null;
    				try {
    					ByteSequence b = (ByteSequence) field.get(bm);
    					by = b.getData();
    				} catch (IllegalArgumentException e1) {
    					e1.printStackTrace();
    				} catch (IllegalAccessException e1) {
    					e1.printStackTrace();
    				}
    				List<String> l = unserializeForList(by);
    				System.out.println(l);
    			}
    		});
    	}
        /**
         * 反序列化列表（用于Redis整存整取）
         * @param in
         * @return
         */
        public static <T> List<T> unserializeForList(byte[] in) {  
            List<T> list = new ArrayList<T>();  
            ByteArrayInputStream bis = null;  
            ObjectInputStream is = null;  
            try {  
                if(in != null) {  
                    bis=new ByteArrayInputStream(in);  
                    is=new ObjectInputStream(bis);  
                    while (true) {  
                        T obj = (T) is.readObject();  
                        if(obj == null){  
                            break;  
                        }else{  
                            list.add(obj);  
                        }  
                    }  
                    is.close();  
                    bis.close();  
                }  
            } catch (IOException e) {  
                log.warn("Caught IOException decoding %d bytes of data",  
                        in == null ? 0 : in.length, e);  
            } catch (ClassNotFoundException e) {  
                log.warn("Caught CNFE decoding %d bytes of data",  
                        in == null ? 0 : in.length, e);  
            } 
            return list;  
        }
    }

三、发送数据的消息类型

//纯字符串的数据
                session.createTextMessage();
                //序列化的对象
                session.createObjectMessage();
                //流，可以用来传递文件等
                session.createStreamMessage();
                //用来传递字节
                session.createBytesMessage();
                //这个方法创建出来的就是一个map，可以把它当作map来用
                session.createMapMessage();
                //这个方法，拿到的是javax.jms.Message，是所有message的接口
                session.createMessage();

（1）发送javabean对象类型，因为需要序列化，所以必须实现Serializable接口

//使用createObjectMessage()，可以发送对象，前提是这个对象需要序列化，因为一切在网络在传输的，都是字节
                ObjectMessage obj = session.createObjectMessage();
                for(int i = 0 ; i < 100 ; i ++){
                    Person p = new Person(i,"名字");
                    obj.setObject(p);
                    producer.send(obj);
                }

（2）接收javabean对象类型

consumer.setMessageListener(new MessageListener() {
                    @Override
                    public void onMessage(Message message) {
                        try {
                            //强转为ObjectMessage，拿到对象后强转为Person
                            Person p = (Person) ((ObjectMessage)message).getObject();
                            System.out.println(p);
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                });

（3）发送文件类型，需要先读取文件，把读取到输入流中的子节数据放到writeBytes()里

//使用createBytesMessage()创建消息
                BytesMessage bb = session.createBytesMessage();
                InputStream is = new FileInputStream("D:/test.txt");
                //开始读取信息
                byte[] b = new byte[5];//把所有的数据读取到这个字节当中
                //声明一个int存储每次读取到的数据
                int i = 0;
                //定义一个记录索引的变量
                int index = 0;
                //循环读取每个数据
                while((i=is.read())!=-1){//把读取的数据放到i中
                    b[index]=(byte) i;
                    index++;
                }
                //把字节数组转成字符串
                System.out.println(new String(b));
                //发送文件到消息中
                bb.writeBytes(new byte[]{2});
                //关闭流
                is.close();

（4）接收文件类型

consumer.setMessageListener(new MessageListener() {
                    @Override
                    public void onMessage(Message message) {
                        BytesMessage bm = (BytesMessage)message;
                        FileOutputStream out = null;
                        try {
                            out = new FileOutputStream("d:/test.txt");
                        } catch (FileNotFoundException e2) {
                            e2.printStackTrace();
                        }
                        byte[] by = new byte[1024];
                        int len = 0 ;
                        try {
                            while((len = bm.readBytes(by))!= -1){
                                out.write(by,0,len);
                            }
                        } catch (JMSException | IOException e1) {
                            e1.printStackTrace();
                        }
                    }
                });

四、确定消息能否成功处理

（1）AUTO\_ACKNOWLEDGE，在之前创建一个session的时候，需要指定其事务，及消息的处理模式，当消息发送给消费者之后，就自动确认成功了，而不管消费者有没有处理成功，而一旦确认成功后，就会把队列里面的消息给清除掉，避免下一个消费者接收到同样的消息

session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

（2）CLIENT\_ACKNOWLEDGE，如果消费者不确认消息，那么activemq将会把这条消息一直保留，直到有一个消费者确定了消息

session = connection.createSession(false, Session.CLIENT_ACKNOWLEDGE);

在消费者接收到消息的时候，调用javax.jms.Message的acknowledge()，这样当消息处理成功之后，手动确认消息，如果不确定，activemq将会发给下一个消费者处理

consumer.setMessageListener(new MessageListener() {
                    @Override
                    public void onMessage(Message message) {
                        try {
                            //获取到接收的数据
                            String text = ((TextMessage)message).getText();
                            System.out.println(text);
                            //确认接收，并成功处理了消息
                            message.acknowledge();
                        } catch (JMSException e) {
                            e.printStackTrace();
                        }
                    }
                });

注：只在Queue队列模式中有效，在Topic主题模式即使不确认，也不会保存消息