Start with communication and talk about message oriented middleware

1、 Communication mode between systems

We can think of ways:

Disadvantages of each mode:

We expect a way of communication:

This is MQ (message queue). MQ meets all the above features. In addition, compared with other communication modes, MQ has the following advantages:

2、 Brief introduction to MQ

From queue to message service

One queue:

Multiple queues:

Two common message modes

Guarantee of message processing

Three QoS

Message ordering

Messages of the same topic or queue are delivered in order

Note: if you perform operations such as message partitioning or batch prefetching, there may be no order

Message protocol

JMS can run on any underlying protocol, but the API is bound to the programming language. AMQP can support two different programming languages to use it to deliver messages.

Why does message oriented middleware not use HTTP protocol?

3、 Message Oriented Middleware

Third generation message middleware:

1. Introduction to ActiveMQ

Main functions:

Languages: Java, C, C + +, c#, ruby, Perl, python, PHP protocols: openwire, stomp rest, WS notification, XMPP, AMQP, mqtt

Usage scenario:

Using tutorials

<dependency>
			<groupId>org.apache.activemq</groupId>
			<artifactId>activemq-core</artifactId>
			<version>5.7.0</version>
 </dependency>

package com.mmc.springbootstudy.activemq.p2p;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

public class Producer {

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,ActiveMQConnection.DEFAULT_PASSWORD,"tcp://0.0.0.0:61616");
        Connection connection = connectionFactory.createConnection();
        connection.start();
        Session session = connection.createSession(false,Session.CLIENT_ACKNowLEDGE);
        Queue queue = session.createQueue("test-queue");
        //生产者
        MessageProducer producer = session.createProducer(queue);
        //设置不持久化
        producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
        //发送消息
        for (int i = 0; i < 1; i++) {
            sendMsg(session,producer,i);
        }
        session.commit();
        connection.close();
    }

    private static void sendMsg(Session session,MessageProducer producer,int i) throws JMSException {
        TextMessage textMessage = session.createTextMessage("Hello ActiveMQ " + i);
        producer.send(textMessage);
    }
}

package com.mmc.springbootstudy.activemq.p2p;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

import javax.jms.*;

/**
 * 接收者
 */
public class JmsReceiver {

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,Session.CLIENT_ACKNowLEDGE);
        Queue queue = session.createQueue("test-queue");
        MessageConsumer consumer = session.createConsumer(queue);
        while (true){
            TextMessage receive = (TextMessage) consumer.receive();
            System.out.println(receive.getText());
            receive.ackNowledge();
        }

    }
}

public class TopSend {

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,Session.CLIENT_ACKNowLEDGE);
        MessageProducer producer = session.createProducer(null);
        TextMessage textMessage = session.createTextMessage("hello,topic");
        Topic topic = session.createTopic("test-topic");
        producer.send(topic,textMessage);
    }
}

public class TopReceiver {

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,Session.CLIENT_ACKNowLEDGE);
        Topic topic = session.createTopic("test-topic");
        MessageConsumer consumer = session.createConsumer(topic);
        while (true){
            TextMessage receive = (TextMessage) consumer.receive();
            System.out.println(receive.getText());
        }

    }
}

Or accept the message by listening.

public class TopReceiver {

    public static void main(String[] args) throws JMSException {
        ConnectionFactory connectionFactory=new ActiveMQConnectionFactory(ActiveMQConnection.DEFAULT_USER,Session.CLIENT_ACKNowLEDGE);
        Topic topic = session.createTopic("test-topic");
        MessageConsumer consumer = session.createConsumer(topic);
        consumer.setMessageListener(new MessageListener() {
            @Override
            public void onMessage(Message message) {
                try {
                    System.out.println(((TextMessage)message).getText());
                } catch (JMSException e) {
                    e.printStackTrace();
                }
            }
        });

    }
}

Unfinished... The next part continues Kafka