标签: jms

我想知道如何根据建议使用 ActiveMQ 进行测试的答案使用 ActiveMQ 测试JMS 模拟 JMS - jUnit

但是，我没有从 MessageConsumer 收到我期望的消息。如果我使用接收它就挂在那里。

这是我的代码

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.junit.Assert;
import org.junit.Test;

public class JmsTest {
@Test
public void test() throws Exception {
    final ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
            "vm://localhost?broker.persistent=true");
    final Queue queue;
    final Connection connection = connectionFactory.createConnection();
    final Session session = connection.createSession(true,
            Session.AUTO_ACKNOWLEDGE);
    {
        queue = session.createQueue("test");
    }
    {
        final MessageProducer producer = session.createProducer(queue);
        final TextMessage message …

我有以下Spring Integration配置.我在这里做的是dequeuing来自主题的消息,并在转换后将其发送到某个HTTP位置.

JMS Connection Factory 配置如下:

<bean id="inboundCF" 
class="org.springframework.jms.connection.CachingConnectionFactory">
    <constructor-arg index="0">
        <jee:jndi-lookup jndi-name="java:comp/resource/ABC_AQ/XATopicConnectionFactories/XATCF" />
    </constructor-arg>
    <property name="sessionCacheSize" value="3" />
</bean>
<bean id="txInboundCF"
class="org.springframework.jms.connection.TransactionAwareConnectionFactoryProxy">
    <property name="targetConnectionFactory" ref="inboundCF" />
<property name="synchedLocalTransactionAllowed" value="true" />
</bean>

并Message Listener Container配置如下:

<bean id="jmsInboundContainer"
class="org.springframework.jms.listener.DefaultMessageListenerContainer"
destroy-method="destroy">
    <property name="connectionFactory" ref="txInboundCF" />
    <property name="destination" ref="inboundDestination" />
    <property name="pubSubDomain" value="true" />
    <property name="sessionTransacted" value="true" />
    <property name="errorHandler" ref="errorHandlerService" />
    <property name="subscriptionDurable" value="true" />
    <property name="durableSubscriptionName" value="mySub" />
    <property name="cacheLevel" value="3" />
</bean>
<int-jms:message-driven-channel-adapter channel="jmsInChannel" …

我正在学习 JMS 并遇到了这个声明：http : //docs.oracle.com/javaee/1.3/jms/tutorial/1_3_1-fcs/doc/advanced.html#1023387

PERSISTENT 传递模式是默认设置，它指示 JMS 提供者格外小心，以确保在 JMS 提供者发生故障时消息不会在传输过程中丢失。使用此传递模式发送的消息在发送时会记录到稳定存储中。

如果发生 JMS Provider 故障，那么 JMS Provider 如何确保消息不丢失？

这是什么意思：

“使用这种传递模式发送的消息在发送时会记录到稳定存储中。”

请帮助我在这里理解 JMS 概念。

我有一个简单的 JMS 客户端 java 代码，它连接到 weblogic 应用程序服务器并将消息发送到定义的 jms 队列。

对于 JMS 客户端与服务器建立的每个连接，在具有 JMS 队列的 Weblogic 应用程序服务器和运行 JMS 客户端代码的机器之间建立一个底层 TCP 连接。

根据我有限的知识，我认为连接中的本地端口是随机选择的。如果这是错误的，请纠正我。

在我们的生产服务器中，由于一些严格的客户政策，我们被要求将本地端口限制为固定端口范围，而不是任何随机端口。是否可以通过在 JMS 客户端代码中指定任何连接工厂属性来做同样的事情。

E.g. 
192.19.81.223 -> m/c where Weblogic is installed
7001 -> Weblogic Server admin port, where the JMS Server is targeted
192.19.105.54 -> m/c where the JMS Client code is running
61372 -> Random port being selected in the m/c where JMS Client is run.

$home > netstat -an|grep 7001
tcp        0      0 ::ffff:192.19.81.223:7001   :::*                             LISTEN
tcp …

我需要编写一个侦听消息队列（消费者）并将其读取的数据发送到数据库的服务器，以及另一个将信息（生产者）发布到该队列的服务器。

我正在关注Spring+JMS 上的官方 spring 参考。

我不明白以下内容：

1. JMS和ActiveMQ有什么关系？
2. 其中哪一个真正实现了消息队列？

3. 在我看到的例子中：

   FileSystemUtils.deleteRecursively(new File("activemq-data"));
   

   Run Code Online (Sandbox Code Playgroud)

   作为删除队列的一种手段。

   队列的数据是否保存在文件中？

4. 如果队列是由文件系统维护的，我如何将我的JMS服务器扩展到多于一台计算机（服务器）

谢谢！

我有一条骆驼路线，如下所示

from("activemq:queue:upload" )
    .pollEnrich().simple("file:basePath/${header.ID}?noop=true&recursive=true")
    .aggregationStrategy(new ExampleAggregationStrategy()) 
    .timeout(2000) 
 .toD("ftp:${header.destinationURI}")

在我的文件系统中file:basePath/${header.ID}包含多个文件夹。执行上述路由时，只会将第一个文件夹中的第一个文件复制到 ftp 服务器。剩余的文件夹（带子文件夹）没有被复制到 ftp 服务器！

而ExampleAggregationStrategy()类的aggregate()方法看起来像以下

@Override
        public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
            String destinationURI = "populatedURI";

        oldExchange.setOut(newExchange.getIn());
        oldExchange.getOut().setHeader("ID",  oldExchange.getIn().getHeader("ID"));
        oldExchange.getOut().setHeader("destinationURI", destinationURI);
        oldExchange.setFromEndpoint(newExchange.getFromEndpoint());
        oldExchange.setFromRouteId(newExchange.getFromRouteId());

            return oldExchange;
        }

我也试过设置properties and onCompletions。仍然没有运气！我错过了什么aggregationStrategy吗？
如何成功复制所有文件和文件夹pollEnrich？

我正在开发一个使用 ActiveMQ 和 Play Framework v2.4.2（Java 版）向最终用户发送电子邮件的消息传递系统。我是 JMS/ActiveMQ 技术的新手。我刚刚在 ActiveMQ 站点上使用了这个 Hello World 示例作为起点。

我创建了一个如下的测试类来测试使用 Play Framework 运行 ActiveMQ，一切正常：

public class ActiveMQMailApp {

    public static void main(String[] args) throws Exception {
        setup();
        MailConsumer.initService();
        for (int i =0;i<11;i++) MailProducer.sendMail(fakeMail());
    }
    public static void setup(){
        FakeApplication fakeApplication = Helpers.fakeApplication();
        Helpers.start(fakeApplication);
    }

    private static Mail fakeMail() throws InterruptedException {
        Thread.sleep(1000);
        SimpleDateFormat sdf = new SimpleDateFormat("YYYY-MM-dd  hh:mm:ss");
        return new Mail( "noreply@abc.com", "receiver@gmail.com", "A Test Email", "<html><body><p>Date: <b> "+sdf.format(new Date())+" </b></p></body></html>");
    }

} …

我得到了一个JmsConfig配置类，它通过以下方式处理来自主题的 JMS 事件：

• 它定义了一个 @Bean ConnectionFactory，包含一个 ActiveMQ 实现
• 它定义了一个@Bean JmsListenerContainerFactory实例化 aDefaultJmsListenerContainerFactory并通过 Boot 的DefaultJmsListenerContainerFactoryConfigurer
• 它定义了一个@Bean MessageConverter包含一个MappingJackson2MessageConverter并设置一个自定义ObjectMapper
• 我@JmsListener在我的服务方法上使用指向 myfactory 的注释。这是我对该主题的唯一用途，仅订阅。

现在我想转到Spring Integration。在阅读了很多之后，如果我不需要双向使用（丢弃网关）也不需要轮询机制（丢弃@InboundChannelAdapter），我将寻求一个message-driven-channel-adapter传统的 XML 配置措辞。我发现 Java 习惯用法应该通过新的 Spring Integration DSL 库来完成，因此，我寻找合适的代码片段。

这似乎JmsMessageDrivenChannelAdapter是正确的等价物，我找到了一种方法：

IntegrationFlows.from(Jms.messageDriverChannelAdapter(...))

但问题是这仅接受 ActiveMQ ConnectionFactory 或AbstractMessageListenerContainer，但没有我的引导预配置JmsListenerContainerFactory！

这应该如何以最终方式实施？

下面是我的生产者配置，如果您看到它们的压缩类型为 gzip ，即使我提到了压缩类型，为什么消息没有发布并且它失败了

props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, edi856KafkaConfig.getBootstrapServersConfig());
        props.put(ProducerConfig.RETRIES_CONFIG, edi856KafkaConfig.getRetriesConfig());
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, edi856KafkaConfig.getBatchSizeConfig());
        props.put(ProducerConfig.LINGER_MS_CONFIG, edi856KafkaConfig.getIntegerMsConfig());
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, edi856KafkaConfig.getBufferMemoryConfig());
        ***props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");***
        props.put(Edi856KafkaProducerConstants.SSL_PROTOCOL, edi856KafkaConfig.getSslProtocol());
        props.put(Edi856KafkaProducerConstants.SECURITY_PROTOCOL, edi856KafkaConfig.getSecurityProtocol());
        props.put(Edi856KafkaProducerConstants.SSL_KEYSTORE_LOCATION, edi856KafkaConfig.getSslKeystoreLocation());
        props.put(Edi856KafkaProducerConstants.SSL_KEYSTORE_PASSWORD, edi856KafkaConfig.getSslKeystorePassword());
        props.put(Edi856KafkaProducerConstants.SSL_TRUSTSTORE_LOCATION, edi856KafkaConfig.getSslTruststoreLocation());
        props.put(Edi856KafkaProducerConstants.SSL_TRUSTSTORE_PASSWORD, edi856KafkaConfig.getSslTruststorePassword());
        **props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "gzip");** 

错误如下

org.apache.kafka.common.errors.RecordTooLargeException: The message is 1170632 bytes when serialized which is larger than the maximum request size you have configured with the max.request.size configuration.
2017-12-07_12:34:10.037 [http-nio-8080-exec-1] ERROR c.tgt.trans.producer.Edi856Producer - Exception while writing mesage to topic= '{}'
org.springframework.kafka.core.KafkaProducerException: Failed to send; nested exception is org.apache.kafka.common.errors.RecordTooLargeException: The message is 1170632 bytes …

我正在尝试使用 java 向 ibmmq 发布一条简单的消息。消息发送成功。但是当我在 ibm 控制台上检查队列时。消息显示为

但我期待作为简单的字符串。

这是我的代码。当我尝试转换时，我收到以下消息 jms_bytes 类型的消息无法将其正文分配给 java.lang.String

import com.ibm.mq.*;
import com.ibm.mq.constants.MQConstants;
import com.ibm.mq.jms.MQQueueConnectionFactory;
import com.ibm.msg.client.jms.JmsFactoryFactory;
import com.ibm.msg.client.wmq.WMQConstants;

import javax.jms.*;
import java.io.IOException;
import java.util.Hashtable;

public class PublisherTest
{
    static private String CHANNEL = "anychannel";
    static private int    PORT = 1414;
    static private String HOST = localhost;
    static private String QMANAGER = "QM1";
    static private String QUEUE = "queue.test";
    static private String USER = USER;
    static private Hashtable<String, Object> props =
            new Hashtable<String, Object>();
    static MQQueueManager qMgr …