hornetq - 在hornetq中，消费者是自动调用的吗？

Question

我查看了 hornetq 中的所有示例，但我找不到当消息通过生产者时自动调用消费者的示例。

请让我知道示例代码或提示。提前致谢。

Answer 1

使用DefaultMessageListenerContainer。您可以为其注册一个侦听器并异步使用消息。请点击此链接了解有关调整 MessageListenerContainer 的更多信息：http: //bsnyderblog.blogspot.se/2010/05/tuning-jms-message-consumption-in.html。

您需要的 Hornetq 依赖项（我使用了独立的 hornetq-2.3.0.CR2）（您还需要一些弹簧罐）：

<dependencies>
    <!-- hornetq -->
    <dependency>
        <groupId>org.jboss.netty</groupId>
        <artifactId>netty</artifactId>
        <version>3.2.7.Final</version>
    </dependency>
    <dependency>
        <groupId>org.hornetq</groupId>
        <artifactId>hornetq-jms-client</artifactId>
        <version>2.3.0.CR2</version>
    </dependency>
    <dependency>
        <groupId>org.hornetq</groupId>
        <artifactId>hornetq-core-client</artifactId>
        <version>2.3.0.CR2</version>
    </dependency>
    <!-- hornetq -->
</dependencies>

您应该在 applicationContext.xml 中使用的 bean（我没有使用 jndi 来获取 ConnectionFactory 和目的地；为此，您可以关注这个问题）：

<!-- It's ConnectionFactory to connect to hornetq. 5445 is hornetq acceptor port  -->
<bean name="connectionFactory" class="messaging.jms.CustomHornetQJMSConnectionFactory"> 
    <constructor-arg index="0" name="ha" value="false" />
    <constructor-arg index="1" name="commaSepratedServerUrls" value="127.0.0.1:5445" /> 
</bean>

<bean id="destinationParent" class="messaging.jms.JmsDestinationFactoryBean" abstract="true">
    <property name="pubSubDomain" value="false" /> <!-- default is queue -->
</bean>

<bean id="exampleDestination" parent="destinationParent">
    <property name="destinationName" value="example" /> <!-- queue name -->
</bean>

 <!-- MessageListener -->
<bean id="messageHandler" class="messaging.consumer.MessageHandler">
</bean>

<!-- MessageListenerContainer -->
    <bean id="paymentListenerContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="destination"       ref="exampleDestination" />
        <property name="messageListener"   ref="messageHandler" />
        <property name="connectionFactory" ref="connectionFactory" />
        <property name="sessionTransacted" value="true" />
        <property name="concurrentConsumers"    value="1" />
        <property name="maxConcurrentConsumers" value="10" />
        <property name="idleConsumerLimit"      value="2" />
        <property name="idleTaskExecutionLimit" value="5" />
        <property name="receiveTimeout"         value="3000" />
    </bean>

CustomHornetQJMSConnectionFactory：

public class CustomHornetQJMSConnectionFactory extends org.hornetq.jms.client.HornetQJMSConnectionFactory
{
    private static final long serialVersionUID = 1L;

    public CustomHornetQJMSConnectionFactory(boolean ha, String commaSepratedServerUrls)
    {
        super(ha, converToTransportConfigurations(commaSepratedServerUrls));
    }

    public static TransportConfiguration[] converToTransportConfigurations(String commaSepratedServerUrls)
    {   
        String [] serverUrls = commaSepratedServerUrls.split(",");
        TransportConfiguration[] transportconfigurations = new TransportConfiguration[serverUrls.length];
        for(int i = 0; i < serverUrls.length; i++)
        {
            String[] urlParts = serverUrls[i].split(":");
            HashMap<String, Object> map = new HashMap<String,Object>();
            map.put(TransportConstants.HOST_PROP_NAME, urlParts[0]);
            map.put(TransportConstants.PORT_PROP_NAME, urlParts[1]);
            transportconfigurations[i] = new TransportConfiguration(NettyConnectorFactory.class.getName(), map);
        }
        return transportconfigurations;
    }
}

JmsDestinationFactoryBean（用于destinationParent bean）：

public class JmsDestinationFactoryBean implements FactoryBean<Destination>
{
    private String destinationName;
    private boolean pubSubDomain = false;

    public void setDestinationName(String destinationName) {
        this.destinationName = destinationName;
    }

    public void setPubSubDomain(boolean pubSubDomain) {
        this.pubSubDomain = pubSubDomain;
    }

    @Override
    public Class<?> getObjectType() 
    {
        return Destination.class;
    }

    @Override
    public boolean isSingleton() 
    {
        return true;
    }

    @Override
    public Destination getObject() throws Exception 
    {
        if(pubSubDomain)
        {
            return HornetQJMSClient.createTopic(destinationName);
        }
        else
        {           
            return HornetQJMSClient.createQueue(destinationName);           
        }
    }
}

MessageHandler（收到的消息转到onMessage方法进行处理）（为简单起见，您可以实现javax.jms.MessageListener而不是SessionAwareMessageListener）：

public class MessageHandler implements org.springframework.jms.listener.SessionAwareMessageListener<Message>
{
@Override
public void onMessage(Message msg, Session session) throws JMSException 
{
    if(msg instanceof TextMessage)
    {
        System.out.println(((TextMessage)msg).getText());
        session.commit();
    }
    else
    {
        session.rollback(); // send message back to the queue
    }
}