3

私は ActiveMQ を JMS ブローカーおよびコンシューマーとして使用し、jmsTemplate を使用してメッセージを送信します。ピーク時には、毎秒約 100 件のメッセージがあります。

キューにメッセージがいくつあるかは関係ありませんが、メッセージが重複することがよくあります。私が思いついた一時的な解決策は、テーブルにインデックスを設定することです。現時点では、すべてのメッセージはデータベースにのみ保存されます。

私の最初の質問 - 非永続トピックを指定し、応答が必要ない場合、メッセージが複製されるのはなぜですか?

送信者:

@Component
public class QueueSender 
{
    private Logger log = Logger.getLogger(getClass());
@Autowired
    protected JmsTemplate jmsTemplate;


    public JmsTemplate getJmsTemplate() {
        return jmsTemplate;
    }

    public void setJmsTemplate(JmsTemplate jmsTemplate) {
        this.jmsTemplate = jmsTemplate;
    }

    @Autowired
    public QueueSender( final JmsTemplate jmsTemplate )
    {
        this.jmsTemplate = jmsTemplate;
        this.jmsTemplate.setDeliveryPersistent(false);
        System.out.println("isSessionTransacted "+jmsTemplate.isSessionTransacted()+
                " getDeliveryMode "+jmsTemplate.getDeliveryMode()+
                " getReceiveTimeout "+jmsTemplate.getReceiveTimeout()+
                " getSessionAcknowledgeMode "+jmsTemplate.getSessionAcknowledgeMode());
    }


    public void sendPrice(Integer tickerId, Integer field, Double price, Long timestamp)
    {
        jmsTemplate.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        jmsTemplate.setMessageIdEnabled(true);
        Map <String, Object>map = new HashMap<String, Object>();
        map.put("tickerId", tickerId);
        map.put("field", field);
        map.put("price", price);
        map.put("timestamp", timestamp);
        jmsTemplate.convertAndSend("Quotez", map);
    }

    public void sendVolume(Integer tickerId, Integer field, Integer size, Long timestamp)
    {
        jmsTemplate.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

        Map <String, Object>map = new HashMap<String, Object>();
        map.put("tickerId", tickerId);
        map.put("field", field);
        map.put("size", size);
        map.put("timestamp", timestamp);
        jmsTemplate.convertAndSend("Quotez", map);

    }

}

リスナー：

public void onMessage(Message message) 
{
     if (message instanceof MapMessage) 
     {           
         try
         {
             MapMessage mapMessage = (MapMessage) message;
                 if(null !=  mapMessage.getString("price"))
                 {
 priceService.insert(mapMessage.getInt("tickerId"),mapMessage.getDouble("price"),
mapMessage.getInt("field"),mapMessage.getLong("timestamp"));
                 }                     else{
volumeService.insert(mapMessage.getInt("tickerId"),mapMessage.getInt("size"),
mapMessage.getInt("field"),mapMessage.getLong("timestamp"));
             }
         }
         catch (final JMSException e)
         {
             exceptionListener.onException(e);
         }
     }
}

春： 

<amq:broker useJmx="true" persistent="false">
<amq:transportConnectors>
  <amq:transportConnector uri="tcp://localhost:0"/>
</amq:transportConnectors> </amq:broker>
<amq:topic id="topicDest"  physicalName="Quotez"/>
  <amq:connectionFactory id="jmsFactory" brokerURL="vm://localhost?jms.watchTopicAdvisories=false"/>  
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
<constructor-arg ref="jmsFactory" />
<property name="exceptionListener" ref="jmsExceptionListener" />
<property name="sessionCacheSize" value="100" />
</bean>


<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
   <constructor-arg ref="connectionFactory"/>
    <property name="pubSubDomain" value="true"/>
 <property name="defaultDestinationName" value="Quotez"/>    
</bean>
<bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="topicDest"/>
        <property name="messageListener" ref="jdbcListener" />
    </bean>

2 番目の質問は、jmsContainer の構成に関するものです。上のコードと下のコードの違いは何ですか? 上記のコードはトピックをサブスクライバーとして提供し、以下のコードはキューを提供します。

<jms:listener-container concurrency="10" connection-factory="connectionFactory">    
<jms:listener id="JdbcListener" destination="topicDest" ref="queueListener" />  
</jms:listener-container>

Camel とそのべき等なConsumer が重複の問題を解決すると思われることがわかりました。3 番目の質問は、Camel の構成に関するものです。私はこの構成を持っています（デフォルト）：

<bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
<property name="brokerURL" value="tcp://localhost:0"/>
</bean>

<bean id="myRepo" class="org.apache.camel.processor.idempotent.MemoryIdempotentRepository"/>

<camelContext xmlns="http://camel.apache.org/schema/spring">
<route>
    <from uri="direct:start"/>
    <idempotentConsumer messageIdRepositoryRef="myRepo">
        <header>messageId</header>
        <to uri="mock:result"/>
    </idempotentConsumer>
</route>
</camelContext>

それはすべてのキューに適用されますか、それとも明示的なサブスクリプションを作成する必要がありますか? すべてのトピック/キューとすべての着信メッセージをチェックすると思います。現時点での問題は、すべてのメッセージに messageId=null があり、フィルターがそれをパラメーターとして受け取ることです。 

2011-03-01 11:24:09,152 DEBUG (org.springframework.jms.core.JmsTemplate:567) - Sending created message: ActiveMQMapMessage {commandId = 0, responseRequired = false, **messageId = null**, originalDestination = null, originalTransactionId = null, producerId = null, destination = null, transactionId = null, expiration = 0, timestamp = 0, arrival = 0, brokerInTime = 0, brokerOutTime = 0, correlationId = null, replyTo = null, persistent = false, type = null, priority = 0, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = null, dataStructure = null, redeliveryCounter = 0, size = 0, properties = null, readOnlyProperties = false, readOnlyBody = false, droppable = false} ActiveMQMapMessage{ theTable = {field=1, timestamp=1298975049138, price=72.89, tickerId=2} }

messageId を設定する簡単な方法が見つかりませんでした。私の質問 - messageId を設定するだけで十分であり、例外として機能するか、構成に問題があります。たとえば、使用するトピックを指定する必要があります。

ありがとう、

ジダス

4

1 に答える 1

4

JMS トピックを使用する場合、同時/最大同時コンシューマーを「1」に設定する必要があります。そうしないと、重複が発生します。マルチスレッドの消費や負荷分散が必要な場合は、代わりに仮想トピックを使用してください。

于 2011-07-08T17:02:07.487 に答える