mq实现代码

这里提供一个Java语言实现的MQ代码示例，基于Apache Kafka：

发送消息代码：

import java.util.Properties;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;

public class MQProducer {
    private String topic;
    private Properties props;
    private Producer<String, String> producer;
    
    public MQProducer(String topic) {
        this.topic = topic;
        this.props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("buffer.memory", 33554432);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        this.producer = new KafkaProducer<>(props);
    }
    
    public void sendMessage(String message) {
        producer.send(new ProducerRecord<>(topic, message));
    }
    
    public void close() {
        producer.close();
    }
}

消费消息代码：

import java.util.Properties;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

public class MQConsumer {
    private String topic;
    private Properties props;
    private Consumer<String, String> consumer;
    
    public MQConsumer(String topic) {
        this.topic = topic;
        this.props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        this.consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(topic));
    }
    
    public String receiveMessage() {
        ConsumerRecords<String, String> records = consumer.poll(1000);
        String message = "";
        for (TopicPartition partition : records.partitions()) {
            List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
            for (ConsumerRecord<String, String> record : partitionRecords) {
                message = record.value();
            }
            long lastOffset = partitionRecords.get(partitionRecords.size() - 1).offset();
            consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(lastOffset + 1)));
        }
        return message;
    }
    
    public void close() {
        consumer.close();
    }
}

以上代码是基于Apache Kafka的实现，你也可以使用其他MQ实现，例如RabbitMQ等。