标签: jms

作为一个简单的例子，我有一个具有以下字段的学说实体

id
name
description

我正在使用JMSSerializerBundle并且它在大多数情况下运行良好，但是如果我想让序列化数据（Json）包含未完全映射到我的实体的内容怎么办？

例如，如果我只想返回描述的前 50 个字符并且我想将其称为short_description.

我尝试使用排除策略来@Expose公开方法，但这不受支持。

我需要经常与许多不同的实体一起做这种事情，我想知道是否有人可以提出一个很好的干净的方法来解决这个问题。

我已经阅读了整个文档，JMSSerializerBundle并在互联网上寻找解决方案，我可以想出一些解决方案，但生成的代码看起来有点脏。

在我的消息队列中包含几条消息。我想按特定属性进行分组。这些都有一个自定义属性“item-id”

我现在将某些价值观定为：体育、电影......

我这样做是这样的：

new MessageCreator() {
 ObjectMessage message = session.createObjectMessage();
 message.setObject(data);
 message.setStringProperty("item-id", "sports");
}

该值也显示在属性的队列中。

当我尝试检索包含此属性的消息时，我没有得到任何结果。

初审：

 Connection con = pc.createConnection();
 Session sess =con.createSession(false,Session.AUTO_ACKNOWLEDGE);       
 AmqMessagesQueryFilter queryFilter = new AmqMessagesQueryFilter(pc, queue); 
 con.start();
 String selector = "item-id = 'sports'";
 List messages = queryFilter.query(selector); 

但列表是空的

二审：

 Connection con = pc.createConnection();
 Session sess = con.createSession(false, Session.AUTO_ACKNOWLEDGE);

 AmqMessagesQueryFilter queryFilter = new AmqMessagesQueryFilter(pc, queue);

 QueueBrowser queueBrowser = sess.createBrowser(queue,"item-id = 'sports'");
 con.start();
 List messages=Collections.list(queueBrowser.getEnumeration());

该列表也是空的。

我的错误是什么？

我缺少什么？

AMQ 版本 5.13.2 Java 1.8.0_74 Windows 10

给定一个简单的测试用例，传输两个对象消息，一个包含数据，另一个是数据结束标记。仅接收数据结束标记。

队列在作业开始时创建，并在作业完成后销毁。

如果我运行大量交易，我会看到大约 50% 的接收率。

日志清楚地显示接收器在第一条消息放入队列之前启动，两条消息都被放入队列，但实际上只接收第二条消息。

发送方和接收方都位于同一个 JVM 上。每个都有自己的会话和连接。

连接和队列设置代码：

@Override
public void beforeJob(JobExecution jobExecution) {
    // TODO Auto-generated method stub
    try {
        jobParameters = jobExecution.getJobParameters();

        readerConnection = connectionFactory.createConnection();
        readerConnection.start();

        writerConnection = connectionFactory.createConnection();
        writerConnection.start();

        jmsQueueManagementSession = writerConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);

        queueName = jobParameters.getString("jobName") + "." + jobExecution.getId();

        queue = jmsQueueManagementSession.createQueue(getQueueName());

    } catch (JMSException ex) {
        throw new MaxisRuntimeException(
                MaxisCodeHelperImpl.generateCode("MXAR", MXMODULE, JMS_RECEIVER_INITIALIZATION_ERROR), null);
    }

}

发送者设置代码：

@Override
public void beforeStep(StepExecution stepExecution) {

    this.stepExecution = stepExecution; …

我正在寻求解决消息传递服务器和队列的 FIFO 性质所遇到的问题。在某些情况下，我希望根据消息传递顺序以外的标准将队列中的消息分发到消费者池。理想情况下，这将防止用户占用系统中的共享资源。以这个过于简化的场景为例：

• 应用程序中有一项功能，用户可以清空垃圾箱。
• 此事件为垃圾桶中的每个项目发送一条 DELETE 消息
• 该队列的使用者调用具有速率限制 API 的 Web 服务。

鉴于每个用户的垃圾箱中可能有大量消息，我们有哪些选项可以允许并发处理每个垃圾箱而不考虑排队时间？在我看来，有几个明显的解决方案：

• 为每个用户创建单独的队列和消费者池
• 将消息从单个队列随机传递到单个消费者池

在我们的例子中，创建一个单独的队列并管理每个用户的消费者确实不切实际。这是可以做到的，但我认为如果合理的话我真的更喜欢第二种选择。我们正在使用 RabbitMQ，但如果有更适合此任务的技术，则不一定与其绑定。

我正在考虑使用 Rabbit 的消息优先级来帮助随机发送的想法。通过随机为消息分配 1 到 10 之间的优先级，这应该有助于分发消息。这种方法的问题是，如果队列永远不会完全清空，那么具有最低优先级的消息可能会永远卡在队列中。我以为我可以在消息上使用 TTL，然后以升级的优先级重新排队消息，但我在文档中注意到了这一点：

应过期的消息仍然只会从队列头部过期。这意味着与普通队列不同，即使每个队列的 TTL 也可能导致过期的低优先级消息卡在未过期的高优先级消息后面。这些消息永远不会被传递，但它们会出现在队列统计信息中。

我担心我可能会因为这种方法而陷入兔子洞。我想知道其他人是如何解决这个问题的。任何有关创意路由、消息传递模式或任何替代解决方案的反馈将不胜感激。

队列管理器可以有多个侦听器吗？到目前为止，我已经使用了一种监听器/队列管理器组合，想知道这是否可能。这是因为我们有 2 个应用程序连接到同一个队列管理器，并且似乎存在问题。

JMS 消息 ID 如下所示ID:10.77.42.209-4280-1477454185311-1:1:1391:1:1。整个字符串被“-”和“:”分成几个部分。

显然，一部分代表生产者的IP地址，一部分可能代表消息ID。但别人有什么办法呢？

所以我的问题是每个部分的含义是什么？

使用ActiveMQ并且没有自定义消息ID

我有一个 JMS 生产者发送两种消息：业务逻辑和心跳消息。目前，两者都由同一个接收器处理，但我现在尝试通过使用选择器为每个接收器提供专用的类。我遇到的问题是，每当我将选择器添加到接收器时，它就会停止接收消息。这是我到目前为止所拥有的。为了简单起见，我只添加了心跳的代码：

要发送消息，我有这个：

private void sendHeartBeat() {
    this.buildTemplate().send(new HeartbeatMessageCreator(this.someId));
}

private JmsTemplate buildTemplate() {
    if (this.cachedJmsTemplate == null) {
        final ActiveMQTopic activeMQTopic = new ActiveMQTopic(this.topic);
        this.cachedJmsTemplate = new JmsTemplate(this.config.getCachedConnectionFactory());
        this.cachedJmsTemplate.setDefaultDestination(activeMQTopic);
        this.cachedJmsTemplate.setPubSubDomain(true);
    }
    return this.cachedJmsTemplate;
}

心跳消息创建者：

class HeartbeatMessageCreator implements MessageCreator {
private final String someID;

HeartbeatMessageCreator(final String someID) {
    this.someID = someID;
}

@Override
public Message createMessage(final Session session) throws JMSException {
    final Serializable message = new ZHeartBeat(this.someID);
    final Message jmsMessage = session.createObjectMessage(message);
    jmsMessage.setJMSType(message.getClass().getName());
    jmsMessage.setStringProperty("InternalMessageType", "HeartBeat"); // <-- Setting …

我已经在我的项目中配置了Spring JMS。现在我的要求是，仅当属性文件中的标志处于 ON 时才读取消息。

在Spring中有哪些方法可以实现这一点？

我在 Spring Boot 中有 2 个 jmsliteners 设置：

@EnableJms

@JmsListener(id= "A1", destination = "dest1", containerFactory = "factory1")
@JmsListener(id= "A2", destination = "dest2", containerFactory = "factory1") 
public void onMessage(String request) {
    //some processing
}

@Bean(name = "factory1")
public JmsListenerContainerFactory jmsListenerContainerFactory(ConnectionFactory cachingConnectionFactory){
 DefaultJmsListenerContainerFactory jmsListenerContainerFactory = new DefaultJmsListenerContainerFactory();
    jmsListenerContainerFactory.setConnectionFactory(cachingConnectionFactory);
    jmsListenerContainerFactory.setConcurrency("2-10");
    jmsListenerContainerFactory.setSessionTransacted(true);
    jmsListenerContainerFactory.setSessionAcknowledgeMode(Session.CLIENT_ACKNOWLEDGE);
    return jmsListenerContainerFactory;
}

由于2个jmslisteners有不同的id但共享相同的containerFactory，这2个jmslisteners是否共享相同的线程池？

即，如果我在 dest1 处有 100 条消息，在 dest2 处有 1 条消息，是否有可能所有可用线程都将被侦听器 a1 占用，并且侦听器 a2 将无法运行，直到 a1 处的大部分消息被消耗为止？

理想情况下，这两个任务应并行运行，我不想在处理中创建一些异步的未来任务。

我想从 jmsTemplate.sendAndReceive 获取同步响应：

    Message responseMessage = producer.produceAndReceive(gzip, mestype, uploadFile.getName(), uploadFile.getAbsolutePath());

它在另一个类中调用 ProduceAndReceive ：

  @Override
  public Message produceAndReceive(final byte[] data, final String type, final String name, final String archivePath) {
    jmsTemplate.setReceiveTimeout(20000);
    return jmsTemplate.sendAndReceive(SAPPI_EXPORT_QUEUE, new MessageCreator() {
      @Override
      public Message createMessage(Session session) throws JMSException {
        String msgId = UUIDGen.getUUID();
        BytesMessage message = session.createBytesMessage();
        message.writeBytes(data);
        message.setStringProperty(ISapProducer.IDOC_TYPE, type);
        message.setStringProperty(ISapProducer.ORIGIN_FILE_NAME, name);
        message.setStringProperty(ISapProducer.MESSAGE_ID, msgId);
        message.setStringProperty(ISapProducer.ARCHIVE_PATH, archivePath);

        message.setJMSReplyTo(session.createTemporaryQueue());
        message.setJMSCorrelationID(msgId);
        return message;
      }
    });
  }

在这一步之后，我认为该消息已经放入队列中。我有 @JmsListener 方法，可以“监听”这个队列：

  @Override
  @JmsListener(destination = "myqueue.export")
  public void consume(final Message message) throws …