ActiveMQ是apache的一个开源消息引擎。可以作为即通引擎或者消息中间件引擎。
准备
下载ActiveMQ
http://activemq.apache.org/download.html
进入\bin\win64双击InstallService.bat安装为系统服务。然后启动这个系统服务
访问 http://localhost:8161/admin/queues.jsp 可以看到消息队列列表 账号密码默认就是admin admin,配置在conf/jetty-realm.properties中
然后将根目录下的activemq-all-5.13.2.jar引入项目，就可以进行开发了
首先对ActiveMQ进行简单封装
1 package activeMQStu; 2 3 import org.apache.activemq.ActiveMQConnectionFactory; 4 5 import javax.jms.*; 6 import java.util.Arrays; 7 8 /** 9 * Created by lvyahui on 2016/4/23. 10 */ 11 public class ActiveMQServer { 12 13 14 private String user; 15 private String pass; 16 private String url; 17 18 private Session session; 19 private Connection connection; 20 private ActiveMQConnectionFactory connectionFactory; 21 22 public ActiveMQConnectionFactory getConnectionFactory() { 23 return connectionFactory; 24 } 25 26 public ActiveMQServer(String user, String pass, String url) throws JMSException { 27 this.user = user; 28 this.pass = pass; 29 this.url = url; 30 31 this.connectionFactory = new ActiveMQConnectionFactory(this.user, this.pass, this.url); 32 /* 33 * 必须指明哪些包的类是可以序列化的 34 * 否则会报错：ActiveMQ Serializable class not available to broker. Reason: ClassNotFoundException 35 * 参考：http://activemq.apache.org/objectmessage.html 36 * */ 37 connectionFactory.setTrustedPackages(Arrays.asList("activeMQStu")); 38 this.connection = connectionFactory.createConnection(); 39 connection.start(); 40 } 41 42 public Session getSession() throws JMSException { 43 if (session == null) { 44 session = connection.createSession(true, Session.AUTO_ACKNOWLEDGE); 45 } 46 return session; 47 } 48 49 public Queue createQueue(String name) throws JMSException { 50 return getSession().createQueue(name); 51 } 52 53 public void close() throws JMSException { 54 if (session != null) { 55 session.commit(); 56 session.close(); 57 } 58 if (connection != null) { 59 connection.close(); 60 } 61 } 62 63 }
消息
1 package activeMQStu; 2 3 import java.io.Serializable; 4 5 /** 6 * Created by lvyahui on 2016/4/23. 7 */ 8 public class User implements Serializable { 9 10 private String username ; 11 private String password ; 12 private String salt ; 13 14 public String getUsername() { 15 return username; 16 } 17 18 public void setUsername(String username) { 19 this.username = username; 20 } 21 22 public String getPassword() { 23 return password; 24 } 25 26 public void setPassword(String password) { 27 this.password = password; 28 } 29 30 public String getSalt() { 31 return salt; 32 } 33 34 public void setSalt(String salt) { 35 this.salt = salt; 36 } 37 38 @Override 39 public String toString() { 40 return "User{" + 41 "username='" + username + '\'' + 42 ", password='" + password + '\'' + 43 ", salt='" + salt + '\'' + 44 '}'; 45 } 46 }
消息生产者
1 package activeMQStu.queue; 2 3 import activeMQStu.ActiveMQServer; 4 import activeMQStu.User; 5 import org.apache.activemq.command.ActiveMQObjectMessage; 6 7 import javax.jms.JMSException; 8 import javax.jms.MessageProducer; 9 import javax.jms.ObjectMessage; 10 import javax.jms.Queue; 11 12 /** 13 * Created by lvyahui on 2016/4/23. 14 */ 15 public class UserProducer { 16 17 private MessageProducer producer; 18 19 public UserProducer(ActiveMQServer activeMQServer, String queueName) throws JMSException { 20 Queue queue = activeMQServer.createQueue(queueName); 21 producer = activeMQServer.getSession().createProducer(queue); 22 } 23 24 public void produce(User user) throws JMSException { 25 ObjectMessage objectMessage = new ActiveMQObjectMessage(); 26 objectMessage.setObject(user); 27 producer.send(objectMessage); 28 } 29 }
发送消息
1 package activeMQStu; 2 3 import activeMQStu.queue.UserProducer; 4 import org.apache.activemq.ActiveMQConnection; 5 6 import javax.jms.JMSException; 7 8 /** 9 * Created by lvyahui on 2016/4/23. 10 */ 11 public class ProduceApp { 12 public static void main(String[] args) throws JMSException { 13 ActiveMQServer activeMQServer = new ActiveMQServer( 14 ActiveMQConnection.DEFAULT_USER, 15 ActiveMQConnection.DEFAULT_PASSWORD, 16 "tcp://localhost:61616" 17 ); 18 UserProducer producer = new UserProducer(activeMQServer,"queue.devlyh"); 19 for(int i =0 ;i < 100;i++){ 20 User user = new User(); 21 user.setUsername("lvyahui".concat(String.valueOf(i))); 22 user.setPassword("admin888" + i); 23 user.setSalt("salt"+i); 24 producer.produce(user); 25 } 26 activeMQServer.close(); 27 } 28 }
运行成功后再页面可以看到有100条消息进入了名字为queue.devlyh的队列中
消息接受者
1 package activeMQStu.queue; 2 3 import activeMQStu.ActiveMQServer; 4 import activeMQStu.User; 5 6 import javax.jms.JMSException; 7 import javax.jms.MessageConsumer; 8 import javax.jms.ObjectMessage; 9 10 /** 11 * Created by lvyahui on 2016/4/23. 12 */ 13 public class UserConsumer { 14 private MessageConsumer consumer; 15 16 public UserConsumer(ActiveMQServer activeMQServer,String queueName) throws JMSException { 17 this.consumer = activeMQServer.getSession() 18 .createConsumer(activeMQServer.createQueue(queueName)); 19 } 20 21 public User consume() throws JMSException { 22 ObjectMessage objectMessage = (ObjectMessage) consumer.receive(); 23 User user = (User) objectMessage.getObject(); 24 return user; 25 } 26 }
接收消息
1 package activeMQStu; 2 3 import activeMQStu.queue.UserConsumer; 4 import org.apache.activemq.ActiveMQConnection; 5 import org.apache.activemq.ActiveMQConnectionFactory; 6 7 import javax.jms.JMSException; 8 import java.util.Arrays; 9 10 /** 11 * Created by lvyahui on 2016/4/23. 12 */ 13 public class ConsumeApp { 14 public static void main(String[] args) throws JMSException { 15 ActiveMQServer activeMQServer = new ActiveMQServer( 16 ActiveMQConnection.DEFAULT_USER, 17 ActiveMQConnection.DEFAULT_PASSWORD, 18 "tcp://localhost:61616" 19 ); 20 21 // ActiveMQConnectionFactory factory = (ActiveMQConnectionFactory) activeMQServer.getConnectionFactory(); 22 23 UserConsumer consumer = new UserConsumer(activeMQServer,"queue.devlyh"); 24 while(true){ 25 User user = consumer.consume(); 26 System.out.println(user); 27 } 28 } 29 }
执行效果如下
