Java如何提高大量数据的处理性能

List<Integer> list = new LinkedList<>();
for (int i = 0; i < 100000; i++) {
    list.add(i);
}

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
int sum = numbers.parallelStream()
                .filter(n -> n % 2 == 0)
                .mapToInt(Integer::intValue)
                .sum();

Cache<String, String> cache = CacheBuilder.newBuilder()
                                           .maximumSize(1000)
                                           .expireAfterWrite(10, TimeUnit.MINUTES)
                                           .build();
cache.put("key", "value");
String value = cache.getIfPresent("key");

Connection conn = DriverManager.getConnection(DB_URL, USER, PASS);
PreparedStatement pstmt = conn.prepareStatement("INSERT INTO table_name (column1, column2) VALUES (?, ?)");
for (int i = 0; i < largeDataSet.size(); i++) {
    pstmt.setInt(1, largeDataSet.get(i).getColumn1());
    pstmt.setString(2, largeDataSet.get(i).getColumn2());
    pstmt.addBatch();
    if (i % 1000 == 0) {
        pstmt.executeBatch(); // 每1000条记录执行一次批量插入
    }
}
pstmt.executeBatch(); // 执行最后一批

import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;
 
public class RocketMQProducer {
    public static void main(String[] args) throws Exception {
        // 创建生产者并指定生产者组名
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroup");
        // 设置NameServer地址
        producer.setNamesrvAddr("localhost:9876");
 
        // 启动生产者
        producer.start();
 
        for (int i = 0; i < 1000; i++) {
            try {
                // 创建消息对象，指定主题、标签和消息体
                Message msg = new Message("TopicTest", "TagA", ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                // 发送消息
                SendResult sendResult = producer.send(msg);
                System.out.printf("%s%n", sendResult);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
 
        // 如果不再发送消息，关闭生产者实例。
        producer.shutdown();
    }
}

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
 
public class RocketMQConsumer {
    public static void main(String[] args) throws Exception {
        // 创建消费者组并指定消费者组名
        DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("ConsumerGroup");
        // 设置NameServer地址
        consumer.setNamesrvAddr("localhost:9876");
 
        // 订阅一个或多个主题来消费消息
        consumer.subscribe("TopicTest", "*");
 
        // 注册回调函数来处理从broker拉取回来的消息
        consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            for (MessageExt msg : msgs) {
                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msg.getBody()));
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
 
        // 启动消费者实例
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}