标签: activemq-classic

我正在尝试对我们的activemq进程进行某种管理,我想知道一些字段意味着我回来了.

特别,

什么是InFlightCount？这是生产者发送的消息数量,而不是测量的是什么是AverageEnqueueTime？秒？什么是EnqueueCount和DequeueCount？什么是DispatchCount？

如果我使用消息队列构建一个包含数十个发布者/订阅者的系统,我似乎有一些网络配置选项:

1. 我可以拥有一台所有机器都使用的集群代理 - 每台机器都没有本地队列
2. 我可以在每台机器上本地安装代理,并使用存储转发将消息传递到远程机器

不同的技术似乎强制执行不同的配置 - 例如,MSMQ要求每台机器都有自己的本地队列,而Tibco EMS似乎经常用于集群,而不是每个消费者都有本地队列.

没有本地队列的缺点是什么,以及影响决策的因素是什么？

所以我有一个ActiveMQ代理和一些生产者,它们在代理中放置了一些自制对象.我也有一些消费者接收这些消息(序列化对象)没有问题.但我想创建一个连接到代理并显示所有消息(序列化对象)的工具.

我尝试使用和ActiveMQConnection这样做:

Set<ActiveMQQueue> currentMessageQueues = activeMQConnection.getDestinationSource().getQueues();
Iterator<ActiveMQQueue> messageQueueIterator = currentMessageQueues.iterator();

while (messageQueueIterator.hasNext()) {
    ActiveMQQueue currentQueue = messageQueueIterator.next();
    QueueSession queueSession = activeMQConnection.createQueueSession(true, Session.AUTO_ACKNOWLEDGE);
    QueueBrowser browser = queueSession.createBrowser(currentQueue);
    Enumeration<?> messagesInQueue = browser.getEnumeration();

    while (messagesInQueue.hasMoreElements()) {
       Message queueMessage = (Message) messagesInQueue.nextElement();

       if (queueMessage instanceof ActiveMQObjectMessage) {
          ActiveMQObjectMessage objectMessage = (ActiveMQObjectMessage) queueMessage;
          objectMessage.getObject();
       }
    }
 }

使用此代码,我在objectMessage.getObject()上得到一个异常:

javax.jms.JMSException: Failed to build body from content. Serializable class not available to broker. Reason: java.lang.ClassNotFoundException: MyMessage

我使用了调试模式,ActiveMQObjectMessage将对象设置为null.

这种方法是否良好,如果是,我做错了什么？我怎样才能从经纪人那里收到这个对象？

我一直在使用嵌入式activeMQ服务器配置(配置非常类似于说明嵌入式activeMQ服务器/监听器概念的示例).作为应用程序的一部分,我有一个监视目录的监视线程.我没有必要终止关闭activeMQ服务器的进程,而是引入了对"STOP"文件的检查,如果文件存在,服务器将关闭.我试图通过调用代理上的stop函数来完成关闭.

(broker.stop())这似乎关闭了服务.但是,在System.exit上,会引发异常错误:

我观察到另一个提到类似行为的帖子的响应表明"这是正常的"行为.另一篇文章指出我应该禁用shutdownhook(通过设置useShutdownHook ="false"来通过代理的Spring配置).这也没有解决问题.

有关此问题的任何帮助表示赞赏.

谢谢

这是异常堆栈跟踪:

17:10:20.056 [Thread-9] DEBUG org.apache.activemq.AdvisoryConsumer - Failed to send remove command: javax.jms.JMSException: Peer (vm://localhost#3) disposed.
javax.jms.JMSException: Peer (vm://localhost#3) disposed.
    at org.apache.activemq.util.JMSExceptionSupport.create(JMSExceptionSupport.java:62) ~[activemq-core-5.5.1.jar:5.5.1]
    at org.apache.activemq.ActiveMQConnection.doAsyncSendPacket(ActiveMQConnection.java:1267) [activemq-core-5.5.1.jar:5.5.1]
    at org.apache.activemq.ActiveMQConnection.asyncSendPacket(ActiveMQConnection.java:1259) [activemq-core-5.5.1.jar:5.5.1]
    at org.apache.activemq.AdvisoryConsumer.dispose(AdvisoryConsumer.java:56) ~[activemq-core-5.5.1.jar:5.5.1]
    at org.apache.activemq.ActiveMQConnection.close(ActiveMQConnection.java:615) [activemq-core-5.5.1.jar:5.5.1]
    at org.springframework.jms.connection.SingleConnectionFactory.closeConnection(SingleConnectionFactory.java:426) [spring-jms-3.1.0.RELEASE.jar:3.1.0.RELEASE]
    at org.springframework.jms.connection.SingleConnectionFactory.resetConnection(SingleConnectionFactory.java:321) [spring-jms-3.1.0.RELEASE.jar:3.1.0.RELEASE]
    at org.springframework.jms.connection.SingleConnectionFactory.destroy(SingleConnectionFactory.java:312) [spring-jms-3.1.0.RELEASE.jar:3.1.0.RELEASE]
    at org.springframework.beans.factory.support.DisposableBeanAdapter.destroy(DisposableBeanAdapter.java:211) [spring-beans-3.1.0.RELEASE.jar:3.1.0.RELEASE]
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.destroyBean(DefaultSingletonBeanRegistry.java:498) [spring-beans-3.1.0.RELEASE.jar:3.1.0.RELEASE]
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.destroySingleton(DefaultSingletonBeanRegistry.java:474) [spring-beans-3.1.0.RELEASE.jar:3.1.0.RELEASE]
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.destroySingletons(DefaultSingletonBeanRegistry.java:442) [spring-beans-3.1.0.RELEASE.jar:3.1.0.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.destroyBeans(AbstractApplicationContext.java:1066) [spring-context-3.1.0.RELEASE.jar:3.1.0.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext.doClose(AbstractApplicationContext.java:1040) [spring-context-3.1.0.RELEASE.jar:3.1.0.RELEASE]
    at org.springframework.context.support.AbstractApplicationContext$1.run(AbstractApplicationContext.java:958) [spring-context-3.1.0.RELEASE.jar:3.1.0.RELEASE]
Caused by: org.apache.activemq.transport.TransportDisposedIOException: Peer (vm://localhost#3) disposed.
    at org.apache.activemq.transport.vm.VMTransport.oneway(VMTransport.java:89) ~[activemq-core-5.5.1.jar:5.5.1]
    at …

我正在对(Java)消息传递和排队系统进行贸易研究,以便即将重新设计主要Web应用程序的后端框架(在Amazon的EC2 Cloud,x-large实例上).我目前正在评估ActiveMQ和RabbitMQ.

计划是有5个不同的队列,其中一个是死信队列.每天发送的邮件数量将介于40K到400K之间.由于我计划将消息内容作为指向数据存储上XML文件位置的指针,因此我希望消息大约为64字节.但是,出于评估目的,我还想考虑在消息中发送原始XML,平均文件大小为3KB.

我的主要问题:每天应该保留多少条消息/多少条消息？考虑到我上面指定的金额,坚持所有消息是否合理？我知道持久性会降低性能,也许会降低很多.但是,由于没有坚持,正在使用大量的RAM.你们有什么建议的？

此外,我知道在线有关于ActiveMQ(JMS)和RabbitMQ(AMQP)的大量信息.我做了大量的研究和测试.看起来这两种实现都符合我的需求.考虑到我上面提供的信息(文件大小和消息数量),任何人都可以指出使用我可能错过的特定供应商的原因吗？

谢谢!

如何在spring上下文中配置多个远程activemq代理(不同的IP地址)？以下是1个远程代理的配置.我使用camel创建路由,从多个远程代理中的不同队列生成和使用消息.根据以下路由,系统如何知道每个队列属于哪个远程代理？

• 项目清单

  从( "直接:启动")来.( "ActiveMQ的:队列:outgoingRequests")

• 项目清单

  from("activemq:queue:incomingOrders").to("log:Events？showAll = true").to("bean:jmsService")

1个代理org.camel.routes的Spring上下文

<bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://10.1.11.97:61616" />
</bean>

<bean id="pooledConnectionFactory"
    class="org.apache.activemq.pool.PooledConnectionFactory" init-
            method="start" destroy-method="stop">
    <property name="maxConnections" value="8" />
    <property name="connectionFactory" ref="jmsConnectionFactory" />
</bean>

<bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
    <property name="connectionFactory" ref="pooledConnectionFactory"/>
    <property name="concurrentConsumers" value="10"/>
</bean>

<bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
    <property name="configuration" ref="jmsConfig"/>
</bean>

我需要通过HTTP/HTTPS-Proxy将ActiveMQ-Listener连接到防火墙外的代理.我到处搜索但没有找到解决方法如何为AcitveMQ-Client设置代理设置.

ActiveMQ正在使用Apache HttpClient,但我不知道如何在ActiveMQ中操纵该客户端的创建.HttpClient不使用htttps.proxyHost和https.proxyPort.

有没有办法为HttpClient的所有实例设置全局http/https代理？

• 我的Java应用程序是部署到JBoss的WAR(4.0.4GA)
• 发布并订阅ActiveMQ(5.6.0)实例
• Java应用程序使用Apache Camel(2.10.3)与ActiveMQ进行所有集成(生成和使用)
• JBoss和ActiveMQ自己(CentOS 5.6 Final)四核虚拟服务器,每个虚拟服务器都在不同的物理上

我有一个线程挂起的问题,我在我的线程转储中看到以下内容:

java.net.SocketInputStream.socketRead0(Native Method)
java.net.SocketInputStream.read(SocketInputStream.java:129)
java.io.BufferedInputStream.fill(BufferedInputStream.java:218)
java.io.BufferedInputStream.read1(BufferedInputStream.java:258)
java.io.BufferedInputStream.read(BufferedInputStream.java:317)
sun.net.www.http.HttpClient.parseHTTPHeader(HttpClient.java:687)
sun.net.www.http.HttpClient.parseHTTP(HttpClient.java:632)
sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1195)
java.net.HttpURLConnection.getResponseCode(HttpURLConnection.java:379)
org.springframework.remoting.httpinvoker.SimpleHttpInvokerRequestExecutor.validateResponse(SimpleHttpInvokerRequestExecutor.java:146)
org.springframework.remoting.httpinvoker.SimpleHttpInvokerRequestExecutor.doExecuteRequest(SimpleHttpInvokerRequestExecutor.java:66)
org.springframework.remoting.httpinvoker.AbstractHttpInvokerRequestExecutor.executeRequest(AbstractHttpInvokerRequestExecutor.java:136)
org.springframework.remoting.httpinvoker.HttpInvokerClientInterceptor.executeRequest(HttpInvokerClientInterceptor.java:192)
org.springframework.remoting.httpinvoker.HttpInvokerClientInterceptor.executeRequest(HttpInvokerClientInterceptor.java:174)
org.springframework.remoting.httpinvoker.HttpInvokerClientInterceptor.invoke(HttpInvokerClientInterceptor.java:142)
org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172)
org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:202)
$Proxy117.SigmaCruxer(Unknown Source)
com.tms.SigmaClient.SigmaClient.processMessage(SigmaClient.java:46)
com.tms.SigmaClient.SigmaServiceEndpoint.doSigma(SigmaServiceEndpoint.java:29)
com.tms.SigmaClient.SigmaServiceEndpoint.mark(SigmaServiceEndpoint.java:43)
sun.reflect.GeneratedMethodAccessor193.invoke(Unknown Source)
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
java.lang.reflect.Method.invoke(Method.java:597)
org.apache.camel.component.bean.MethodInfo.invoke(MethodInfo.java:329)
org.apache.camel.component.bean.MethodInfo$1.proceed(MethodInfo.java:231)
org.apache.camel.component.bean.BeanProcessor.process(BeanProcessor.java:169)
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:104)
org.apache.camel.component.bean.BeanProcessor.process(BeanProcessor.java:74)
org.apache.camel.impl.ProcessorEndpoint.onExchange(ProcessorEndpoint.java:102)
org.apache.camel.impl.ProcessorEndpoint$1.process(ProcessorEndpoint.java:72)
org.apache.camel.impl.converter.AsyncProcessorTypeConverter$ProcessorToAsyncProcessorBridge.process(AsyncProcessorTypeConverter.java:50)
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:78)
org.apache.camel.processor.SendProcessor$2.doInAsyncProducer(SendProcessor.java:114)
org.apache.camel.impl.ProducerCache.doInAsyncProducer(ProducerCache.java:284)
org.apache.camel.processor.SendProcessor.process(SendProcessor.java:109)
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:78)
org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:98)
org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:89)
org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:69)
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:78)
org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:98)
org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:89)
org.apache.camel.processor.interceptor.TraceInterceptor.process(TraceInterceptor.java:99)
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:78)
org.apache.camel.processor.RedeliveryErrorHandler.processErrorHandler(RedeliveryErrorHandler.java:318)
org.apache.camel.processor.RedeliveryErrorHandler.process(RedeliveryErrorHandler.java:209)
org.apache.camel.processor.DefaultChannel.process(DefaultChannel.java:305)
org.apache.camel.processor.UnitOfWorkProcessor.process(UnitOfWorkProcessor.java:102)
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:78)
org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:98)
org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:89)
org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:69)
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:104)
org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:85)
org.apache.camel.component.jms.EndpointMessageListener.onMessage(EndpointMessageListener.java:91)
org.springframework.jms.listener.AbstractMessageListenerContainer.doInvokeListener(AbstractMessageListenerContainer.java:560)
org.springframework.jms.listener.AbstractMessageListenerContainer.invokeListener(AbstractMessageListenerContainer.java:498)
org.springframework.jms.listener.AbstractMessageListenerContainer.doExecuteListener(AbstractMessageListenerContainer.java:467)
org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:325)
org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:263)
org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1058)
org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1050)
org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:947)
java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
java.lang.Thread.run(Thread.java:662)

根据这两个技术:( …

我正试图ActiveMQ 5.8.0在我的项目中使用.有两种不同的存储配置,KahaDB和LevelDB.根据问题,Kaha可以比Level更快或Level可以比Kaha更快.

它们之间真正的区别是什么？

我有一个单一的Java应用程序,围绕@Service我的业务服务层的Spring bean 构建.通常,我的每个业务服务方法都有Spring Security注释(例如@PreAuthorize)来为该操作强制执行适当的授权规则.

在主Web应用程序流程中,这非常有效; 每个Web请求隐式地具有由会话cookie等处理的认证.

但是,当涉及到与其他"内部"系统的各种集成点时,我没有看到清晰的解决方案.

例如,我将从JMS队列中消耗方法,该队列已经在代理中定义了自己的身份验证和授权规则,因此我想隐含地"信任"我得到的消息.但是,就像现在的情况一样,一个简单的Camel路线如下:

WidgetService widgetService = lookup(WidgetService.class);
from("activemq:newWidget")
    .unmarshall(...)
    .bean(widgetService, "newWidget");

最后扔了一个AuthenticationCredentialsNotFoundException.

这告诉我Camel正在正确地调用我的bean,并且从Spring应用了所有神奇的AOP.

对于其他类似的事情,我已经围绕系统的入口点(例如,围绕Quartz Job的execute方法)应用AOP建议,这注入了一个PreAuthenticatedAuthenticationToken,但我不确定这是否真的是最好的方法.

我是否应该继续在建议中包含这些"可信"入口点以添加身份验证上下文,或者我是否应该更改我的服务层以使某些业务方法的特殊形式不需要身份验证,并确保我明确记录他们不是用于网络@Controller方法等？