为JMS消息创建分配ParallelStream处理

Question

我想要缩放下面的示例程序，该程序创建一个简单的JMS消息并同时将其推送到相同的队列。示例程序目前在单个四核主机上需要大约20秒。任何人都可以推荐一些更改以改进性能？ 20秒的测量只是为下面的并行流处理线：

我能想到的

test.parallelStream().forEach(e -> sender.sendMessage(e)); 

一种方法是我的专辑（“测试”）展开了一些主机，然后同时处理集合中的块通过配置每个线程拥有自己的线程池。其中一个缺点是容错性，并且必须放置适当的构造以确保每个线程池不处理相同的消息。

另一种方法是使用更高性能/并行经纪人像卡夫卡

请注意，我采取任何异步线程的办法，我确实需要能够控制多少消息正在从这些线程发送到另一个应用程序，因为它们只同时支持一定数量。任何其他想法？

全部来源：

import java.util.ArrayList; 
import java.util.List; 

import javax.jms.Connection; 
import javax.jms.ConnectionFactory; 
import javax.jms.Destination; 
import javax.jms.JMSException; 
import javax.jms.MessageProducer; 
import javax.jms.Session; 
import javax.jms.TextMessage; 
import org.apache.activemq.ActiveMQConnection; 
import org.apache.activemq.ActiveMQConnectionFactory; 

public class Sender { 
    private static ConnectionFactory factory = null; 
    private static Connection connection = null; 
    private static Session session = null; 
    private static Destination destination = null; 
    private static MessageProducer producer = null; 

    public Sender() {} 

    public void sendMessage(String test) { 
     try { 
      TextMessage message = session.createTextMessage(); 
      message.setText(test); 
      producer.send(message); 
     } catch (JMSException e) { 
      e.printStackTrace(); 
     } 
    } 

    public static void main(String[] args) throws JMSException { 
      factory = new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_BROKER_URL); 
    ((ActiveMQConnectionFactory)factory).setUseAsyncSend(true); 
    ((ActiveMQConnectionFactory)factory).setOptimizeAcknowledge(true); 
    connection = factory.createConnection(); 
    connection.start(); 
    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); 
    destination = session.createQueue("SAMPLEQUEUE"); 
    producer = session.createProducer(destination); 
    producer.setDeliveryMode(DeliveryMode.PERSISTENT); 

     List<String> test = new ArrayList<String>(); 
     for (int i = 0; i <= 100000; i++) { 
      test.add(Integer.toString(i)); 
     } 
     Sender sender = new Sender(); 
     test.parallelStream().forEach(e -> sender.sendMessage(e)); 
    } 
} 

Answer 1

正如指出的MessagProducer不能保证是线程安全的，但也不是会话。

反正就一点：

1. 我会强烈建议不要使用IO操作的默认流API，因为它使用的所有操作一个线程池，线程池被限制为核心的数量另外你不能指定超时。您应该使用的Executor来代替：https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newCachedThreadPool--

 

    ExecutorService executor = Executors.newFixedThreadPool(2); 
    for (String msg: msgs) { 
     executor.execute(() -> send(msg)); 
    } 

如果没有保证的消息传递，您可以更改消息传递模式NON_PERSISTENT，这加快了发送，因为有没有消息存储和保证传送的开销。

在现实中，虽然你不需要自己做线程管理作为ActiveMQ的有异步支持，这意味着你的消息可以在单独的线程，只要承认你不必保证交货：http://activemq.apache.org/async-sends.html

如果确实需要为了保证传送，请改用JMS事务内部的PERSISTENT消息传送模式（默认）批量发送消息（一种全有或全无的方式）。这会提高性能，因为您只在提交期间进行实际发送。 https://docs.oracle.com/javaee/6/api/javax/jms/Connection.html#createSession(boolean, int)

启用optimizeAcknowledge在ActiveMQ中加快消息代理本身：http://activemq.apache.org/optimized-acknowledgement.html

最后，您可以设置会话确认模式Session.DUPS_OK_ACKNOWLEDGE，这样的消息在后台懒洋洋地承认，但是这可能产生重复发送邮件，因此在消费者需要使用唯一的编号或类似编号

当然，您不应该将所有这些方法结合在一起使用，只需使用适合其他方法的常识即可。

Answer 2

除了上面所说的，我还会专注于如何使用这些消息。如果您的接收端应用程序可以同时处理有限数量的消息，则无论您同时发送多少条消息都无所谓，因为它们都将在队列中等待，直到接收应用程序准备好消耗它们。所以，说你的接收应用程序可以同时处理10个请求。我会在您的队列中设置10个消费者，每个消费者只有在处理之前处理的请求之后才从队列中读取传入请求。这样你传入消息的速度并不重要（除了担心队列溢出）。而您的应用程序将始终同时处理不超过10个请求。