Shulou(Shulou.com)06/01 Report--

Editor to share with you how Java Spring boot integrates RabbitMQ to achieve B2B2C Mini Program e-commerce. I hope you will get something after reading this article. Let's discuss it together.

Previously we sent and received messages in the queue, and now it's time to introduce a full messaging mode in RabbitMQ.

Let's take a quick look at what we learned earlier:

Producer: a program that sends messages

Queue (queue): buffer for storing messages

Consumer (consumer): programs that receive messages

The core idea of the RabbitMQ message model is that the producer does not send any messages directly to the queue. In fact, the producer doesn't even know if the message has been delivered to the queue.

The producer only needs to send the message to a switch (exchange). The switch is very simple. It receives messages from the publisher and pushes them to the queue. The switch must know how to handle the messages it receives, whether it should be pushed to the specified queue or multiple queues, or simply ignore the message. These rules are defined by switch type (exchange type).

There are several switch types to choose from: direct, topic, headers and fanout. Here we focus on the last one-- fanout. Fanout exchange is simple, and you can probably guess from the name that it sends messages to all the queues it knows. This is exactly what we need.

First create a switch of type fanout, named "tut.fanout"

@ Profile ({"tut3", "pub-sub"}) @ Configurationpublic class Tut3Config {/ * define an Exchange * * @ return FanoutExchange * / @ Bean public FanoutExchange queue () {return new FanoutExchange ("tut.fanout") Configuration of consumer side: queues, bindings * / @ Profile ("receiver") private static class ReceiverConfig {@ Bean public Queue autoDeleteQueue1 () {return new AnonymousQueue ();} @ Bean public Queue autoDeleteQueue2 () {return new AnonymousQueue () } @ Bean public Binding binding1 (FanoutExchange fanout, Queue autoDeleteQueue1) {return BindingBuilder.bind (autoDeleteQueue1) .to (fanout);} @ Bean public Binding binding2 (FanoutExchange fanout, Queue autoDeleteQueue2) {return BindingBuilder.bind (autoDeleteQueue2) .to (fanout);} @ Bean public Tut3Receiver receiver () {return new Tut3Receiver () @ Bean @ Profile ("sender") public Tut3Sender sender () {return new Tut3Sender ();}}

As with the previous two tutorials, we have defined two profiles (tut3, pub-sub) to ensure that we can run the specified example. A consumer-side configuration is then created, in which two AnonymousQueue and two Binding are defined to bind the corresponding queue to the switch.

Switch list

Rabbitmqctl can list all the switches on the server. There are some switches called amq.* in this list. These are created by default, but you don't need to use them at this time.

$sudo rabbitmqctl list_exchanges

Listing exchanges...

Logs fanout

Amq.direct direct

Amq.topic topic

Amq.fanout fanout

Amq.headers headers

... Done.

Unnamed switch (Nameless exchange)

Previously, we knew nothing about switching, but we were still able to send messages to the queue. What's going on? Because we use the default switch, it is identified by an empty string ("").

We used to send messages like this:

Template.convertAndSend (queue.getName (), message)

The specific source code is as follows. We can see that we used the switch defined by the default null character before.

/ RabbitTemplate@Overridepublic void convertAndSend (String routingKey, final Object object) throws AmqpException {convertAndSend (this.exchange, routingKey, object, (CorrelationData) null);} private volatile String exchange = DEFAULT_EXCHANGE;/\ * _ Alias for amq.direct default exchange. _ / private static final String DEFAULT_EXCHANGE = ""

Temporary queue (Temporary queues)

We used queues with specified names (hello and work-queues). It's important for us to be able to name queues-we need to point the worker process to the same queue. When we want to share the queue between producers and consumers, it is important to give the queue a name.

But this is not the case with our logger. We want to log all log messages, not just some of them. We are only interested in the current news, not the old news. In order to solve this problem, we need to do two things.

First of all, every time we connect to RabbitMQ, we need a new empty queue. We can create a queue with a random name, or it's best to have the server choose a random queue name for us.

Second, when disconnected from the consumer (consumer), the queue should be deleted immediately.

In Spring AMQP, we can use AnonymousQueue as a temporary queue. It is a non-persistent, exclusive queue that can be deleted automatically.

Beanpublic Queue autoDeleteQueue1 () {return new AnonymousQueue ();} @ Beanpublic Queue autoDeleteQueue2 () {return new AnonymousQueue ();}

At this point our queue name will look like this: amq.gen-JzTY20BRgKO-HjmUJj0wLg

Bind (Bindings)

Now that we have created a fanout exchange and two queues, we need to tell the clearinghouse to send messages to our queue. The relationship between the switch and the queue is called binding. In the Tut3Config above, you can see that we have two bindings, one for each AnonymousQueue.

@ Beanpublic Binding binding1 (FanoutExchange fanout, Queue autoDeleteQueue1) {return BindingBuilder.bind (autoDeleteQueue1) .to (fanout);}

Binding list

You can use rabbitmqctl list_bindings to list all existing bindings.

Rabbitmqctl list_bindings

Code integration

Producer

The producer program that sends the message is not much different from the previous one. The most important change is that we are now posting messages to our fanout exchange, rather than the default switch. We need to provide a routingKey when sending, but for fanout exchange, this value will be ignored.

Public class Tut3Sender {@ Autowirdprivate AmqpTemplate template;@Overrideprivate FanoutExchange fanout; private int dots = 0; private int count = 0 (fixedDelay = 1000, initialDelay = 500) public void send () {StringBuilder builder = new StringBuilder ("Hello"); if (dots++ = = 3) {dots = 1;} for (int I = 0; I < dots; iTunes +) {builder.append ('.') } builder.append (Integer.toString (+ + count)); String message = builder.toString (); template.convertAndSend (fanout.getName (), ", message); System.out.println (" [x] Sent'"+ message +");}}

Description:

Prohibit publishing to a switch that does not exist

If no queue is bound to the switch, the message will be lost

Consumers' understanding of the springcloud architecture can be added: 3536247259

We define two consumers to monitor two queues.

Public class Tut3Receiver {@ RabbitListener (queues = "# {autoDeleteQueue1.name}") public void receiver1 (String in) throws InterruptedException {receive (in, 1);} @ RabbitListener (queues = "# {autoDeleteQueue2.name}") public void receiver2 (String in) throws InterruptedException {receive (in, 2);} private void receive (String in, int instance) throws InterruptedException {StopWatch watch = new StopWatch (); watch.start () System.out.println ("instance" + instance + "[x] Received'" + in + "'"); doWork (in); watch.stop (); System.out.println ("instance" + instance + "[x] Done in" + watch.getTotalTimeSeconds () + "s") } private void doWork (String in) throws InterruptedException {for (char ch: in.toCharArray ()) {if (ch = ='.') {Thread.sleep (1000);}

Running

Maven compilation

Mvn clean package-Dmaven.test.skip=true

Run (it is recommended to run the consumer first, then the producer)

Java-jar target/rabbitmq-tutorial-0.0.1-SNAPSHOT.jar-spring.profiles.active=tut3,receiver-tutorial.client.duration=60000

Java-jar target/rabbitmq-tutorial-0.0.1-SNAPSHOT.jar-spring.profiles.active=tut3,sender-tutorial.client.duration=60000

Output

/ / Sender

Ready... Running for 60000ms

[X] Sent 'Hello.1'

[X] Sent 'Hello... 2'

[X] Sent 'Hello... 3'

/ / Receiver

Ready... Running for 60000ms

Instance 2 [x] Received 'Hello.1'

Instance 1 [x] Received 'Hello.1'

Instance 2 [x] Done in 1.002s

Instance 1 [x] Done in 1.002s

Instance 2 [x] Received 'Hello... 2'

Instance 1 [x] Received 'Hello... 2'

Instance 2 [x] Done in 2.003s

Instance 1 [x] Done in 2.003s

Instance 1 [x] Received 'Hello... 3'

Instance 2 [x] Received 'Hello... 3'

Instance 1 [x] Done in 3.011s

Instance 2 [x] Done in 3.011s

After reading this article, I believe you have a certain understanding of "how Java Spring boot integrates RabbitMQ to achieve B2B2C Mini Program e-commerce". If you want to know more about it, welcome to follow the industry information channel, thank you for reading!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.