标签: jms

假设您有一个JMS队列,并且多个消费者正在查看队列中的消息.您希望其中一个消费者获得所有特定类型的消息,因此您决定使用消息选择器.

例如,您可以在名为的JMS消息头中定义一个属性targetConsumer.您的消息选择器,您应用于称为的消费者A,就像是WHERE targetConsumer = 'CONSUMER_A'.

很明显,消费者A现在只是抓住具有属性集的消息,就像在示例中一样.但是,其他消费者是否会意识到这一点？IOW,CONSUMER_A如果它在消费者之前查看队列,那么另一个不受消息选择器限制的消费者会抓取消息A吗？我是否需要将消息选择器应用于WHERE targetConsumer <> 'CONSUMER_A'其他人？

我现在正在RTFMing并收集经验数据,但希望有人可能知道他们的头脑.

请让我知道有什么区别:

• 消息驱动Bean(MDB)
• Java消息服务(JMS)

我在使用hornetQ的ubuntu上使用jboss AS 6 Final

我使用管理面板在名为Message Buffer Queue的服务器上创建了一个新的Queue.

我收到以下错误:

Unable to validate user: guest for check type CONSUME for address jms.queue.MessageBufferQueue

这是我的文件:

package org.jboss.ejb3timers.example;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Enumeration;
import java.util.Hashtable;
import java.util.UUID;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.QueueBrowser;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.QueueSession;
import javax.jms.Session;
import javax.naming.Context;
import javax.naming.InitialContext;

public class TestClass {

    ConnectionFactory Hconnection=null;
    Queue  q=null;
    Connection Hconn=null;
    Context lContext=null;
    MessageConsumer messageConsumer=null;
    MessageProducer messageProducer=null;
    javax.jms.Session session=null;

    /** …

完成以下任务的最佳方法是什么？

• @MessageDriven bean在数据库上做了一些工作
• 失败时,我想回滚数据库事务
• 但是我也希望JMS消息不被重新传递,即不要重新尝试.

我能想到的几个方法可能会工作.还有其他什么,哪个最好？

• 使用@TransactionManagement(type=BEAN)和UserTransaction,并在捕获异常后显式回滚.例如:

  catch (Exception e) { e.printStackTrace(); utx.rollback(); }

• 使用容器管理的事务,指定@TransactionAttribute(value=NOT_SUPPORTED)上onMessage然后委托DB活动与一个单独的方法@TransactionAttribute(value=REQUIRED).

• 保留事务处理,并在服务器中重新配置重试属性.我正在使用Glassfish 3.1.1,我不确定如何设置它.

• 保留所有内容并明确检查消息是否在正文中重新发送,onMessage如果重新发送则退出.(message.getJMSRedelivered()？)

什么运作良好？有没有一种标准/最佳实践方法来处理这个问题？

所以问题是如何使用以下方式临时停止并启动使用spring创建的jms侦听器:

<amq:connectionFactory id="exampleJmsFactory" brokerURL="tcp://${jms.broker.url}" />

<jms:listener-container concurrency="1" connection-factory="exampleJmsFactory"  destination-type="queue" message-converter="exampleMessageConverter">
        <jms:listener destination="incoming.example.client.queue" ref="exampleProductsMessageConsumer" method="consume"/>
</jms:listener-container>


<bean id="exampleProductsMessageConsumer" class="com.unic.example.jms.receive.JmsExampleProductsMessageConsumer" scope="tenant"/>

基本上问题是什么呢.我们确实有一个初始化/更新机制,客户端可以随时运行,并且在这个初始化/更新期间我想停止消耗任何消息,因为系统在这段时间内不可用,如果有消息,它将丢失.

所以我如何使用API​​来停止监听器或监听器容器或整个连接.我发现一个类AbstractJmsListeningContainer有停止/启动但我怎么能得到它？我的意思是没有这个jms:listener和listener-containers有一个名字或类似的东西.

有没有办法设置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++; …

我写了一个Java程序来连接到Websphere MQ来发布消息.我在Websphere MQ Explore中创建了JNDI名称空间,连接工厂,目标和队列管理器.当我运行我的程序时,它显示ClassCastException从类型转换string到ConnectionFactory.

这是我的代码.任何人都可以帮助解决此问题.

JNDIUtil.java

package com.tradefinance.jms.util;

//JMS classes
import javax.jms.JMSException;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;

//JNDI classes
import javax.naming.InitialContext;
import javax.naming.Context;
import javax.naming.NamingException;

//Standard Java classes
import java.util.Hashtable;
import java.util.Properties;

/**
* 
* A wrapper class for JNDI calls
*
*/
public class JNDIUtil
{
    private Context context;

    public JNDIUtil(String icf, String url) throws JMSException, NamingException
    {
        Hashtable environment = new Hashtable();

        environment.put(Context.INITIAL_CONTEXT_FACTORY, icf ); 
        environment.put(Context.PROVIDER_URL, url);

     context= new InitialContext( environment );

    } …

我正在尝试通过FOSRestBundle和JMS Serializer将带有ManyToOne链接的JSON对象('Module')返回到Sonata\MediaBundle\Entity.我应该怎么做呢？

这是我做过的黑客攻击,但不认为这是最好的做法.

class Module
{
    ...
    /**
     * @var Application\Sonata\MediaBundle\Entity\Media
     *
     * @ORM\ManyToOne(targetEntity="Application\Sonata\MediaBundle\Entity\Media",  inversedBy="module")
     * @ORM\JoinColumn(name="hero_image_id", referencedColumnName="id")
     * @JMS\Expose()
     */
    private $heroImage;
    ...
}

class Media extends BaseMedia
{
   ...
   /**
     * A Quick hack not the best method.
     *
     * @JMS\VirtualProperty
     * @JMS\SerializedName("url")
     *
     * @return string
     */
    public function getUrlMethod()
    {
        global $kernel;

        $imageProvider = $kernel->getContainer()->get('sonata.media.provider.image');

        return $imageProvider->generatePublicUrl($this, 'reference');
    }
    ...
}

谢谢!

编辑 感谢Tautrimas Pajarskas和他提到的帖子.这是班级.

<?php
namespace AXO\APIBundle\Listener\Serialization;

use JMS\DiExtraBundle\Annotation\Service;
use JMS\DiExtraBundle\Annotation\Tag;
use JMS\DiExtraBundle\Annotation\Inject;
use JMS\DiExtraBundle\Annotation\InjectParams; …

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