|
该帖已经被评为良好帖
|
|
|---|---|
| 作者 | 正文 |
|
发表时间：2009-05-21
chinaway 写道 博主你好，
我怎么在http://127.0.0.1:8161/admin/topics.jsp中看不到JMS-TEST-TOPIC消息传递的效果呢，消息数量和名称等都看不到 附加本例子所需要的jar包 activemq-all-5.2.0.jar commons-logging-1.1.jar log4j-1.2.14.jar spring2.5.5.jar xbean-spring-2.8.jar(2.8以上) jdk jee 这些jar包都可以在activemq的解压包中找到 给个贴图看看 |
| 返回顶楼 | |
|
发表时间：2009-05-22
谢谢博主，可能是我运行的有问题，或者对ActiveMQ理解不够，只能看到消息发送、接收在控制台的输出，如果不单独启动ActiveMQ,连http://localhost:8161/admin都无法访问。
我的理解是Spring整合ActiveMQ后，运行TestMain后，ActiveMQ的broker会自动启动，但是为什么不能通过http://localhost:8161/admin访问。 |
| 返回顶楼 | |
|
发表时间：2009-05-22
谢谢博主，我的问题解决了，要想在http://127.0.0.1:8161/admin/queues.jsp中看到，必须把<!-- embedded ActiveMQ Broker -->
<amq:broker useJmx="false" persistent="true"> <amq:persistenceAdapter> <amq:amqPersistenceAdapter directory="d:/amq"/> </amq:persistenceAdapter> <amq:transportConnectors> <amq:transportConnector uri="tcp://localhost:61616" /> <amq:transportConnector uri="vm://localhost:0" /> </amq:transportConnectors> </amq:broker> 这一段中的<amq:transportConnectors> <amq:transportConnector uri="tcp://localhost:61616" /> <amq:transportConnector uri="vm://localhost:0" /> </amq:transportConnectors>注释掉，即使用外部的ActiveMQ服务 |
| 返回顶楼 | |
|
发表时间：2009-07-27
最后修改：2009-07-27
我的消息每次都会重复被接受，这是为什么呢？
结果如下 13:43:43,921 INFO [STDOUT] ****************** Topic B : ID001BUG管理需求和设计. doc 13:43:43,921 INFO [STDOUT] ****************** Topic B : ID001BUG管理需求和设计. doc 13:43:43,921 INFO [STDOUT] ************** Topic A : ID001BUG管理需求和设计.doc 13:43:43,921 INFO [STDOUT] ************** Topic A : ID001BUG管理需求和设计.doc =====================spring配置文件如下============================
<beans
xmlns="http://www.springframework.org/schema/beans"
xmlns:amq="http://activemq.org/config/1.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd http://activemq.org/config/1.0 http://activemq.apache.org/schema/core/activemq-core-5.0-SNAPSHOT.xsd" >
<!--
推荐版本，使用spring的listenerContainer，消息用数据库持久化保存，服务器重启不会丢失
-->
<!-- embedded ActiveMQ Broker -->
<amq:broker useJmx="false" persistent="true">
<amq:persistenceAdapter>
<amq:jdbcPersistenceAdapter id="jdbcAdapter" dataSource="#dataSource" createTablesOnStartup="true"
useDatabaseLock="false"/>
<!--
Mysql can setup useDatabaseLock="true",this is defualt
HSQLDB,MSSQL plz setup useDatabaseLock="false",
if u setup useDatabaseLock="true",u will catch error:
MSSQL Error Info:FOR UPDATE clause allowed only for DECLARE CURSOR
HSQLDB Error Info:FOR in statement [SELECT * FROM ACTIVEMQ_LOCK FOR UPDATE]
see http://www.nabble.com/ActiveMQ-JDBC-Persistence-with-SQL-Server-tf2022248.html#a5560296
-->
</amq:persistenceAdapter>
<amq:transportConnectors>
<amq:transportConnector uri="tcp://localhost:0"/>
</amq:transportConnectors>
</amq:broker>
<!-- 连接外部的activeMQ
<amq:connectionFactory id="jmsConnectionFactory" brokerURL="tcp://localhost:61616?wireFormat.maxInactivityDuration=0" />
ActiveMQ connectionFactory -->
<!-- ActiveMQ connectionFactory 连接内部的-->
<amq:connectionFactory id="jmsConnectionFactory" brokerURL="vm://localhost"/>
<!-- ActiveMQ destinations 使用Queue方式 -->
<amq:queue name="destination" physicalName="org.apache.activemq.spring.Test.spring.embedded"/>
<!-- 使用topic方式-->
<amq:topic name="TOPIC" physicalName="JMS-TEST-TOPIC" />
<!-- The msSQL Datasource that will be used by the Broker
<bean id="mssql-ds" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close">
<property name="driverClassName" value="net.sourceforge.jtds.jdbc.Driver"/>
<property name="url">
<value>jdbc:jtds:sqlserver://localhost:1433/wmjqxgl;SelectMethod=cursor;charset=GBK;tds=8.0;lastupdatecount=true</value>
</property>
<property name="username" value="sa"/>
<property name="password" value="sa"/>
<property name="poolPreparedStatements" value="true"/>
</bean>
-->
<!-- Spring JmsTemplate config -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
<property name="connectionFactory">
<!-- lets wrap in a pool to avoid creating a connection per send -->
<bean class="org.springframework.jms.connection.SingleConnectionFactory">
<property name="targetConnectionFactory" ref="jmsConnectionFactory"/>
</bean>
</property>
<!-- custom MessageConverter -->
<property name="messageConverter" ref="orderMessageConverter"/>
</bean>
<!-- OrderMessage converter -->
<bean id="orderMessageConverter" class="com.work.activemq.OrderMessageConverter"/>
<!-- POJO which send Message uses Spring JmsTemplate -->
<bean id="orderMessageProducer" class="com.work.activemq.OrderMessageProducer">
<property name="template" ref="jmsTemplate"/>
<property name="destination" ref="destination"/>
</bean>
<!-- topic 方式信息发送者 -->
<bean id="topicMessageProducer" class="com.work.activemq.TopicMessageProducer">
<property name="template" ref="jmsTemplate" />
<property name="destination" ref="TOPIC" />
</bean>
<!-- consumer1 for topic a 消息消费者 -->
<bean id="topicConsumerA" class="com.work.activemq.TopicConsumerA" />
<!-- consumer2 for topic a -->
<bean id="topicConsumerB" class="com.work.activemq.TopicConsumerB" />
<!-- Message Listener for -->
<bean id="topicListenerA" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="topicConsumerA" />
<!-- may be other method -->
<property name="defaultListenerMethod" value="receive" />
<!-- custom MessageConverter define -->
<property name="messageConverter" ref="orderMessageConverter" />
</bean>
<bean id="topicListenerB" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg ref="topicConsumerB" />
<!-- may be other method -->
<property name="defaultListenerMethod" value="receive" />
<!-- custom MessageConverter define -->
<property name="messageConverter" ref="orderMessageConverter" />
</bean>
<!-- Message Driven POJO (MDP) 通过queue的方式发送消息，一个发送一个接收-->
<bean id="messageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
<constructor-arg>
<bean class="com.work.activemq.OrderMessageConsumer">
<!-- <property name="mailService" ref="mailService"/> -->
</bean>
</constructor-arg>
<!-- may be other method -->
<property name="defaultListenerMethod" value="sendEmail"/>
<!-- custom MessageConverter define -->
<property name="messageConverter" ref="orderMessageConverter"/>
</bean>
<!-- listener container，MDP无需实现接口 -->
<bean id="listenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="jmsConnectionFactory"/>
<property name="destination" ref="destination"/>
<property name="messageListener" ref="messageListener"/>
</bean>
<!-- listener container，MDP无需实现接口 -->
<bean id="topicListenerContainerA" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="jmsConnectionFactory" />
<property name="destination" ref="TOPIC" />
<property name="messageListener" ref="topicListenerA" />
</bean>
<bean id="topicListenerContainerB" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
<property name="connectionFactory" ref="jmsConnectionFactory" />
<property name="destination" ref="TOPIC" />
<property name="messageListener" ref="topicListenerB" />
</bean>
<bean id="orderNotify" class="com.work.activemq.OrderNotifyImpl">
<property name="orderMessageProducer" ref="orderMessageProducer" />
<property name="topicMessageProducer" ref="topicMessageProducer" />
</bean>
<!-- -->
</beans>
================== |
| 返回顶楼 | |
|
发表时间：2009-08-17
TO: wmj2003
把你的java代码也贴出来看看 |
| 返回顶楼 | |
|
发表时间：2009-08-17
package com.work.activemq;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.HashMap;
import java.util.Map;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import org.apache.activemq.command.ActiveMQObjectMessage;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.jms.support.converter.MessageConverter;
/**
* 订单消息转换类. 使用数据库持久化消息的时候使用。
* <p>实现MessageConverter使得JMS的发送与接收者可以直接发送POJO而不是JMS Message.</p>
*
* @author wangmj
* @see MessageConverter
*/
public class OrderMessageConverter implements MessageConverter {
private static final Log log = LogFactory.getLog(OrderMessageConverter.class);
/*
* (non-Javadoc)
*
* @see org.springframework.jms.support.converter.MessageConverter#toMessage(java.lang.Object,
* javax.jms.Session)
*/
public Message toMessage(Object obj, Session session) throws JMSException {
//check Type
if (obj instanceof Order) {
ActiveMQObjectMessage objMsg = (ActiveMQObjectMessage) session.createObjectMessage();
HashMap<String, byte[]> map = new HashMap<String, byte[]>();
try{
// POJO must implements Seralizable
ByteArrayOutputStream bos = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(bos);
oos.writeObject(obj);
map.put("Order", bos.toByteArray());
objMsg.setObjectProperty("Map", map);
} catch (IOException e) {
log.error("toMessage(Object, Session)", e);
}
return objMsg;
} else {
throw new JMSException("Object:[" + obj + "] is not Order");
}
}
/*
* (non-Javadoc)
*
* @see org.springframework.jms.support.converter.MessageConverter#fromMessage(javax.jms.Message)
*/
@SuppressWarnings("unchecked")
public Object fromMessage(Message msg) throws JMSException {
if (msg instanceof ObjectMessage) {
HashMap<String, byte[]> map = (HashMap<String, byte[]>) ((ObjectMessage) msg).getObjectProperty("Map");
try {
// POJO must implements Seralizable
ByteArrayInputStream bis=new ByteArrayInputStream((byte[])map.get("Order") );
ObjectInputStream ois = new ObjectInputStream(bis);
Object returnObject = ois.readObject();
return returnObject;
} catch (IOException e) {
log.error("fromMessage(Message)", e);
} catch (ClassNotFoundException e) {
log.error("fromMessage(Message)", e);
}
return null;
} else {
throw new JMSException("Msg:[" + msg + "] is not Map");
}
}
}
package com.work.activemq;
import javax.jms.Topic;
import org.springframework.jms.core.JmsTemplate;
/**
* 奇怪啊，topic方式，消息被重复消费了。每次都消费两次。
* 一旦更换为DefaultMessageConverter.java ，控制台就看不到打印信息了。<br>
* 但是在activemq的监控界面，可以看到消息被消费了，而是web发送信息页面也提示成功。
* 估计是和Serializable有关系。
* @author wangmingjie
* @date 2009-7-26上午11:21:32
*/
public class TopicMessageProducer {
private JmsTemplate template;
private Topic destination;
public void setTemplate(JmsTemplate template) {
this.template = template;
}
public void setDestination(Topic destination) {
this.destination = destination;
}
/**
* 发送信息
* @param message
*/
public void send(Order message) {
template.convertAndSend(this.destination, message);
}
}
package com.work.activemq;
/**
* @author wangmingjie
* @date 2009-7-26上午11:27:00
*/
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 使用多线程接受消息
*/
public class TopicConsumerA {
ExecutorService exec = Executors.newFixedThreadPool(10);
public void receive(final Order message) {
//System.out.println("线程名称"+Thread.currentThread().getName());
exec.submit((new Runnable() {
public void run() {
//stem.out.println(Thread.currentThread().getName());
System.out.println("**** Topic A : " + message.getId()+message.getName());
}
}));
}
}
package com.work.activemq;
/**
* @author wangmingjie
* @date 2009-7-26上午11:27:05
*/
public class TopicConsumerB {
public void receive(Order message) {
System.out.println("**** Topic B : " + message.getId()+message.getName());
}
}
|
| 返回顶楼 | |
|
发表时间：2009-08-17
最后修改：2009-08-17
package com.work.activemq;
public interface OrderNotifyInterface {
/**
* 通知用户订单已经发送！生成者接口
* @param order
*/
public void notifyOrder(Order order);
/**
* 通过topic的方式发送信息
* @param order
*/
public void notifyTopic(Order order);
}
实现如下：
package com.work.activemq;
/**
* 用来测试mq的。
* @author wangmingjie
*
*/
public class OrderNotifyImpl implements OrderNotifyInterface {
private OrderMessageProducer orderMessageProducer;
public void setOrderMessageProducer(OrderMessageProducer orderMessageProducer) {
this.orderMessageProducer = orderMessageProducer;
}
private TopicMessageProducer topicMessageProducer;
public void setTopicMessageProducer(TopicMessageProducer topicMessageProducer) {
this.topicMessageProducer = topicMessageProducer;
}
/**
* 向顾客的邮箱发送订单通知,使用JMS发送.
*/
public void notifyOrder(Order order) {
orderMessageProducer.send(order);
}
public void notifyTopic(Order order){
topicMessageProducer.send(order);
}
}
|
| 返回顶楼 | |
