标签: jms

我有一个涉及发送JMS消息的过程.该过程是交易的一部分.如果事务的后续部分失败,发送消息的前一部分之后的部分,我需要取消该消息.有一种想法是以某种方式设置消息,它不会被捡起一段时间,如果我需要回滚,那么我可以去取消消息.不知道消息,我不知道这个想法是否可行.或者,有更好的主意吗？谢谢

这个问题

如何配置ActiveMQ和<flow>进入Mule ESB 3.2,以确保从队列中提取的消息最终由外部正确处理CXF service？

脚本

我有一个CXF端点,它应该接收传入消息并尽快将其传输到三个外部服务.我们称它们为EX1,EX2,EX3.由于<all>Mule 3.x中引入了组件,这非常简单.

整个解决方案最重要的要求是确保每个收到的消息最终都交付给所有三个CXF服务.所以我们最终得出了这个想法,将每个传入的消息放入Persistent JMS queues(Q1,Q2,Q3).在从队列Qn读取消息之后,它直接转移到相应的EXn端点,因此 - 外部服务.

配置

(我可以根据要求提供完整的配置)

我们已经按照此处的描述配置了ActiveMQ代理,并将其与我们的<flow>配置连接起来.一切似乎按预期工作,我有JConsole连接到我的应用程序,所以我可以看到消息是PERSISTENT类型,他们最终排队等候.如果一切顺利 - 所有三个服务EXn都会收到消息.

测试

当我们关闭其中一个服务时,问题就出现了,让我们说EX2,并重新启动整个服务器模拟失败.消息最终会丢失(我想这不是那么持久,是吧？).最奇怪的是 - 如果我们在EX2关闭时发送了10条消息,那么在服务器重启后,其中9条正在被重新传送!所以我想也许,也许,这10条消息中有9条已经正确入队,而当服务器发生故障时,这条消息中的9条经常被重新传送.

这让我觉得,CXF端点没有处理事务支持,说实话我无法理解.毕竟,当我尝试重新传递时,我可以看到消息在队列中,所以它应该被保留.这显然不是,但为什么呢？

我自己的尝试 我已经尝试过很多东西,但都没有.总是有一条消息丢失.

1. 不要<jms:transaction />在流程中使用任何标签 - 不起作用
2. 在收到消息时启动jms事务,在发送时加入 <cxf:jaxws-client />
3. 将XA与JBoss一起使用<xa-transaction />- 并没有用
4. 提供<default-exception-strategy>配置 - 如果我记得它让事情变得更糟

任何帮助表示赞赏,谢谢.

CONFIG

ACTIVE MQ CONFIGURATION

<spring:bean id="AmqDefaultPolicyEntry" class="org.apache.activemq.broker.region.policy.PolicyEntry">
    <spring:property name="queue" value="queue.*"/>
    <spring:property name="deadLetterStrategy" ref="AmqDeadLetterStrategy"/>
</spring:bean>

<spring:bean …

我要求设计一个通知/发布,其子系统发送电子邮件的应用程序.我计划使用jms发布/订阅(主题)消息来做到这一点

现在的可见性,将有20到30个订阅者,并且将要发布的消息数量将在每天30000到50000个消息的范围内.

我计划使用ActiveMQ JMS + Spring 3 + Tomcat 6实现 问题

1. 我是JMS的新手,我想知道上面的负载是否很高？

2. 我们真的需要在服务器上部署单独的ActiveMQ,还是我们在Webapp中使用嵌入式ActiveMQ就足够了？

3. 单独的ActiveMQ服务器/嵌入式服务器有哪些优点/缺点？

有没有办法设置Amazon SQS队列或Amazon SNS主题才能作为本机XA JMS资源适配器工作？

我们的想法是用SQS替换大型应用程序中activemq的使用,但避免不必要的代码重写开销.

当消息可见性超时时,自动中止当前的XA事务将是很好的.

我列出了每个队列中的队列和消息.以下是我的代码.但是,QueueBrowser无法正确检索邮件.

比方说,我有一个名为TestQueue的队列,它有1000条消息.

我第一次运行我的程序时它只显示200条消息.第二 - 400第三 - 600第四 - 800第五 - 1000

你能告诉我如何解决这个问题吗？

ConnectionFactory out = new ActiveMQConnectionFactory("tcp://localhost:61616?jms.prefetchPolicy.all=10000");
ActiveMQConnection connection = (ActiveMQConnection) out.createConnection();

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

Set<ActiveMQQueue> amqs = connection.getDestinationSource().getQueues(); 
Iterator<ActiveMQQueue> queues = amqs.iterator(); 

while ( queues.hasNext() ) 
{ 
    ActiveMQQueue queue_t = aqueues.next(); 
    String q_name = queue_t.getPhysicalName();
    List<ActiveMQMessage> msgList = ((ActiveMQSession) session).getUnconsumedMessages();

    System.out.println( "\nQueue = " + q_name);

    QueueBrowser queueBrowser = session.createBrowser(queue_t);
    Enumeration e = queueBrowser.getEnumeration();

    int numMsgs = 0;
    while(e.hasMoreElements()) 
    {
         Message message = (Message) e.nextElement();
         numMsgs++; …

我在JMS上运行测试用例,发现处理是顺序的.当我向一个使用JMS发送消息的servlet发出200个请求时,接收者(messageListner)正在按顺序恢复请求.如何接收并发请求？我们要设置任何参数吗？我阅读了JMS教程和API,它们在同一个会话中消息是按顺序传递的,即使我正在为每个发送请求创建一个新会话,并且在接收端仍然有10个会话仍处理顺序.

public class ProducerServlet extends javax.servlet.http.HttpServlet implements
    javax.servlet.Servlet {

// Defines the JNDI context factory.
public final static String JNDI_FACTORY = "weblogic.jndi.WLInitialContextFactory";

// Defines the JMS context factory.
public final static String JMS_FACTORY = "jms/TestConnectionFactory";

// Defines the queue.
public final static String QUEUE = "jms/TestJMSQueue";

public final static String TOPIC = "jms/TestTopic";

TestJMSListener jms = new TestJMSListener();
ConnectionFactory connectionFactory = null;
Queue dest1 = null;
Topic dest =null;
Connection connection = null;
MessageProducer producer = null;

protected void doGet(HttpServletRequest …

什么是JMSType用于？它可以用于定义消息有效负载吗？例如,有效负载可能用于添加产品,JMSType可能是AddProduct.

我在使用Apache CAMEL实现Websphere MQ(WMQ)连接器时遇到了很多困难,该连接器可以无异常地处理MQ确认传递(CoD)报告,也不会以不需要的响应数据报的形式出现副作用.最后,我按照我想要的方式工作,如果您习惯于编写本机MQ客户端,那么这是一种非常标准和通用的方式.我在同一主题的帖子中记录了该方法,但我发现解决方案臃肿而复杂,并且非常感谢任何建议或示例,以使实现更清晰,更优雅.

我理解这个问题的根源在于MQ设计请求 - 回复消息交换模式(MEP)的方式,而不是JMS规范的方式,而不是其JMS组件中的请求 - 应答MEP的CAMEL实现.三种不同的哲学!

1. WMQ具有MessageType标头(请参阅MQMD字段和常量),其中请求值为1,回复为2,数据报为8(单向MEP).此外,值4用于以CoD(交付配置),PAN(肯定确认)和NAN(否定确认)的形式标记报告消息,其中 - 在消息流方面 - 也进行额外的回复信息.可以使用另一个名为"Report"的标题字段为Request消息,回复或数据报请求CoD,PAN和NAN ack,其中可以组合所有报告变体的标志.其他标头字段'ReplyToQ'和'ReplyToQMgr'指定原始发件人期望报告和回复的队列和队列管理器,以及固定的24字节'CorrelId'字段 - 可选 - 可以帮助将报告和回复与原始数据报或请求消息相关联.为了使其更复杂,实际上可以使用相同的原始消息ID和没有CorrelID发回回复和报告,或者在CorrelId中提供原始消息ID,或者在原始请求或数据报中已经指定时返回CorrelId值.IBM 通过WMQ提供JMS API,允许通过WMQ将纯JMS交换作为传输(在额外消息头名称MQRFH2的帮助下),或将本机MQ消息映射到JMS消息,反之亦然.
2. 另一方面,JMS规范提供了一个可选的'JMSReplyTo'头字段和一个'JMSCorrelationID',但是确实将MEP语义保留给客户端应用程序; 即在规格中陈述:" 答复可能是可选的;由客户自行决定. "
3. CAMEL以XML或Java DSL中的"路由"和内部Exchange对象模型为特征,目的是支持EIP模式,其中包括Request-Reply模式.然后CAMEL在其JMS组件中假定如果设置了JMSReplyTo字段,则这必然是期望回复的请求,导致Exchange的Out部分(或者如果Out为空,则部分修改)返回到JMSReplyTo中定义的队列. .

我愿意通过远程Websphere队列管理器支持本地MQ消息与交付确认(CoD)报告的交换,这样除了事务和持久性(即没有丢失,没有重复)之外,我还可以跟踪消息何时被消费并在发生延误时提出警报.

入境问题:

默认情况下,Websphere队列管理器在队列中的消息消耗完成时生成CoD报告.因此,在没有任何特定设置的情况下,当CAMEL端点消耗消息时,远程MQ客户端发送带有CoD标志的数据报(以及当时强制的ReplyToQ)将从队列管理器获得第一个回复作为MQ报告,然后是第二个(意外的)回复消息由CAMEL显式返回并包含CAMEL路由末尾的Exchange对象中剩余的内容,因为CAMEL假定存在JMSReplyTo字段的请求 - 应答EIP(从MQ ReplyToQ和ReplyToQMgr映射)被要求支持CoD回流).

出境问题:

如果没有特定设置,CAMEL默认也会在出站连接上采用请求 - 回复EIP/MEP.然后,CAMEL JMS/MQ端点将等待1个响应.当OUTbound消息是MQ上的JMS(因此具有MQRFH2头)时,这可以正常工作.当强制普通的vanilla MQ,即删除下面的MQRFH2头时,我无法使端点侦听器与相关的传入MQ报告匹配,尽管跟踪值看起来都是正确的(强制执行24个char相关ID以便截断更长的CorrelId值或空填充通过MQ无法对相关过滤器进行地理化处理).有没有人能够解决这个问题？

详细信息:虽然IBM JMS API接受传递特定的JMS属性值WMQ_MESSAGE_BODY = {1 | 0}/WMQ_TARGET_CLIENT = {1 | 0}来控制生成的消息中是否存在JMS头MQRFH2,但这些选项通过CAMEL变得不可用.必须使用CamelJmsDestinationName标头(如CAMEL JMS …

我使用初始化程序创建了一个Spring Boot项目,我正在尝试创建我的第一条消息,但我不知道从哪里开始.我熟悉使用JEE的相同过程,所以我想我需要创建一个工厂,一个发件人和一个消费者.

有人可以帮我吗？

Camel中的wireTap和多播的默认行为似乎相似.这两者都有助于以不同的方式处理相同的消息.那么wireTap和组播之间的主要区别是什么？