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

This article mainly introduces "how to achieve Redis publish subscription". In daily operation, I believe many people have doubts about how to realize Redis publish subscription. The editor consulted all kinds of materials and sorted out simple and easy-to-use operation methods. I hope it will be helpful for you to answer the doubts about "how to achieve Redis publish subscription". Next, please follow the editor to study!

Suppose we have a business scenario in which we need to notify the inventory service for shipping after the order is paid by the website.

It is not difficult to implement the above business. We just need to provide the inventory service to the relevant supplier, and after issuing the order payment, we only need to call the inventory service.

If there is a new business, such as a points service, he needs to get the result of the payment for the order, and then increase the user's points.

This implementation is not difficult, let the points service also provide an interface, after the order payment is made, just call the inventory service.

If you need to get the payment result of issuing an order for two businesses, that's good, and the program changes quickly. However, with the continuous development of business, more and more new business is said to be the result of order payment.

At this point, we will find that there are many problems with the above system architecture:

First, the order payment business is heavily coupled with other businesses. Whenever a new business needs to pay results, it needs to change the order payment business.

Second, if too many services are called, the response time of the order payment interface will be longer. In addition, if the response of any downstream interface becomes slower, synchronization will cause the order payment interface response to become longer.

Third, if any downstream interface fails, it may lead to data inconsistency. For example, in the following figure, first call A, then B after success, and finally call C.

If an exception occurs when calling API B, it may cause the return of the order payment API to fail, but at this time, API A has actually been called successfully, which means that it has handled the result of successful payment for the order issued internally.

This will lead to the three downstream interfaces of A Magi BMague C, An acquires the payment result successfully, but BMagne C does not get it, resulting in the inconsistency of the data of the three systems.

In fact, let's think carefully, for the order payment business, it does not need to care about the result of downstream calls, as long as there is a mechanism to notify them.

At this point, it is necessary to introduce the publish and subscribe mechanism today.

Redis publish and subscribe

Redis provides a messaging mechanism based on the publish / subscribe pattern, in which there is no need for direct communication between message publishers and subscribers.

As shown in the figure above, the message publisher only needs to publish the message on the specified channel, and every client that subscribes to that channel can receive the message.

Using the Redis publish and subscribe mechanism, for the above business, the order payment business only needs to send a message to the payment result channel, and other downstream businesses subscribe to the payment result channel, they can receive the corresponding message, and then make business processing.

In this way, the call relationship between the upstream and downstream of the system can be decoupled.

Let's take a look at how to use the Redis publish and subscribe feature.

Redis provides a set of commands that can be used to publish messages, subscribe to channels, unsubscribe, and subscribe according to patterns.

First of all, let's take a look at how to publish a message, which is actually very simple by using the publish instruction:

Publish channel message

In the figure above, we use the publish instruction to send a message to the channel pay_result. We can see that redis returns 0 to us, which actually represents the number of current subscribers. Since there are no subscriptions at this time, the return result is 0.

Next we use subscribe to subscribe to one or more channels

Subscribe channel [channel...]

As shown in the picture above, we subscribe to the pay_result channel when other clients send messages to this channel

Current subscribers will receive a message.

Our son is using subscription commands, and we need a few main points:

First, after the client executes the subscription instruction, it enters the subscription state, and then it can only receive subscribe, psubscribe, unsubscribe, and punsubscribe commands.

Second, new subscribers will not be able to receive previous messages from this channel because Redis does not persist published messages.

Compared with many professional MQ, such as kafka and rocketmq, the redis publish and subscribe function is a bit crude. However, the redis publish and subscribe feature is better than simplicity, and you can choose to use it if the current scenario can tolerate these shortcomings.

In addition to the above functions, Redis also supports pattern matching subscriptions. To put it simply, the client can subscribe to a pattern with a * sign, and if the names of certain channels match this pattern, then when other clients send messages to those channels, the clients that subscribe to this pattern will also receive the message.

Using the Redis subscription mode, we need to use a new directive psubscribe.

We execute the following instruction:

Psubscribe pay.*

Then we can receive messages once other clients go to channels at the beginning of pay, such as pay_result and pay_xxx.

If we need to unsubscribe from the mode, we need to use the appropriate punsubscribe directive, such as the unsubscribing mode above:

The use of punsubscribe pay.*Redis client publish / subscribe Development publish / subscribe based on Jedis

After talking about the Redis publish and subscribe instructions, let's take a look at how Java Redis clients use publish subscriptions.

The following example is mainly based on the Jedis,maven version:

Redis.clients jedis 3.1.0

Other Redis clients are more or less the same.

It's easy for jedis to distribute the code, as long as you call the publish method of the Jedis class.

/ / do not use this in production environment. It is recommended to use JedisPool thread pool Jedis jedis = new Jedis ("localhost", 6379); jedis.auth ("xxxxx"); jedis.publish ("pay_result", "hello world")

The subscription code is relatively complex. We need to inherit the relevant methods in the JedisPubSub implementation. Once another client sends a message to the subscribed channel, the corresponding method of JedisPubSub will be called.

Private static class MyListener extends JedisPubSub {@ Override public void onMessage (String channel, String message) {System.out.println ("receive subscription channel:" + channel + "message:" + message);} @ Override public void onPMessage (String pattern, String channel, String message) {System.out.println ("receive specific subscription channel:" + channel + "subscription model:" + pattern + "message:" + message);}} "

Second, we need to call the subscribe method of the Jedis class:

Jedis jedis = new Jedis ("localhost", 6379); jedis.auth ("xxx"); jedis.subscribe (new MyListener (), "pay_result")

When another client sends a message to the pay_result channel, the subscription receives the message.

It is important to note, however, that jedis#subscribe is a blocking method, and the call will block the main thread, so if you need to use an asynchronous thread to run in a formal project, the specific code will not be demonstrated here.

Develop publish and subscribe based on Spring-Data-Redis

Native jedis publish and subscribe operations are still relatively complex. Now many of our applications have been based on SpringBoot development, using spring-boot-starter-data-redis, we can simplify publish and subscribe development.

First, we need to introduce the corresponding startter dependencies:

Org.springframework.boot spring-boot-starter-data-redis lettuce-core io.lettuce redis.clients jedis

Here we use Jedis as the underlying connection client, so we need to exclude lettuce and then introduce Jedis dependencies.

Then we need to create a message receiving class that needs a method to consume the message:

Slf4jpublic class Receiver {private AtomicInteger counter = new AtomicInteger (); public void receiveMessage (String message) {log.info ("Received"); counter.incrementAndGet ();} public int getCount () {return counter.get ();}}

Then we just need to inject Spring- Redis-related Bean, such as:

StringRedisTemplate, used to manipulate the Redis command

MessageListenerAdapter, the message listener, can create a message acceptance class Receiver on top of this class injection

RedisConnectionFactory, create the Redis underlying connection

@ Configurationpublic class MessageConfiguration {@ Bean RedisMessageListenerContainer container (RedisConnectionFactory connectionFactory, MessageListenerAdapter listenerAdapter) {RedisMessageListenerContainer container = new RedisMessageListenerContainer (); container.setConnectionFactory (connectionFactory); / / subscribe to specified channels using ChannelTopic / / subscription mode using PatternTopic container.addMessageListener (listenerAdapter, new ChannelTopic ("pay_result"); return container } @ Bean MessageListenerAdapter listenerAdapter (Receiver receiver) {/ / inject Receiver, specify the acceptance method return new MessageListenerAdapter (receiver, "receiveMessage") in the class;} @ Bean Receiver receiver () {return new Receiver ();} @ Bean StringRedisTemplate template (RedisConnectionFactory connectionFactory) {return new StringRedisTemplate (connectionFactory);}}

Finally, we use StringRedisTemplate#convertAndSend to send a message, and Receiver will receive a message.

@ SpringBootApplicationpublic class MessagingRedisApplication {public static void main (String [] args) throws InterruptedException {ApplicationContext ctx = SpringApplication.run (MessagingRedisApplication.class, args); StringRedisTemplate template = ctx.getBean (StringRedisTemplate.class); Receiver receiver = ctx.getBean (Receiver.class); while (receiver.getCount () = = 0) {template.convertAndSend ("pay_result", "Hello from Redis!"); Thread.sleep (500L) } System.exit (0);}}

Redis publish and subscribe actual application Redis Sentinel node discovery

Redis Sentinel is a high availability scheme for Redis, which can automatically promote the slave node to the master node when the master node fails, thus transferring the fault.

Instead of explaining the detailed principles of Redis Sentinel today, let's take a look at how Redis Sentinel uses the publish and subscribe mechanism.

Redis Sentinel nodes mainly use publish and subscribe mechanism to discover new nodes and exchange states between primary nodes.

As shown below, each Sentinel node will periodically send messages to the _ sentinel_:hello channel, and each Sentinel will subscribe to this node.

In this way, once a node sends a message to this channel, other nodes can receive the message immediately.

In this way, once a new node joins, it sends a message to this channel, and after other nodes receive it, it determines that the local list does not have this node, so it can join the local node list as a new node.

In addition, the content of each message sent to this channel can contain the status information of the node, which can be used as a basis for the subsequent election of Sentinel leaders.

The above is for the Redis server, for the client, we can also use the publish and subscribe mechanism.

When Redis Sentinel fails over the master node, the various stages of the process are made available through publish subscriptions.

As far as our client is concerned, we are more concerned about the master node after the switch, so that we can switch the connection of the master node in time (the old node has failed at this time and can no longer accept operation instructions)

The client can subscribe to the + switch-master channel, and once the Redis Sentinel ends the failover of the master node, it will publish the message of the master node.

Redission distributed lock

The redission open source framework provides some convenient ways to manipulate Redis, among which the well-known redission implements distributed locks based on Redis.

Today we will take a look at how to use the Redis publish and subscribe mechanism in Redis's implementation of distributed locks to improve locking performance.

For the principle of PS:redission distributed lock implementation, you can refer to the previous article:

Implementation of reentrant distributed lock

Redis distributed lock seems simple, but in fact it is not.

First, let's take a look at how redission locks:

Redisson redisson =.... RLock redissonLock = redisson.getLock ("xxxx"); redissonLock.lock ()

RLock inherits from the Lock interface of the Java standard and calls the lock method. If the current lock has been acquired by another client, the current locked thread will be blocked until the other client releases the lock.

There is actually a question here: how do currently blocked threads perceive that distributed locks have been released?

There are actually two ways to do this:

First, regularly query the status of the lock when it is distributed, and add the lock once it is found that the lock has been released (this key value does not exist in the Redis).

The pseudo code is implemented as follows:

While (true) {boolean result=lock (); if (! result) {Thread.sleep (N);}}

This approach is easy to implement, but it also has many disadvantages.

If the scheduled task time is too short, it will lead to too many queries, in fact, these are invalid queries.

If the scheduled task dormant time is too long, it will lead to locking time is too long, resulting in poor locking performance.

Then the second implementation scheme is to use the service notification mechanism, when the distributed lock is released, the client can receive the lock release message, and then add the lock as soon as possible.

We can use the Redis publish and subscribe model for this service notification mechanism.

When a thread fails to lock, the thread will subscribe to the redisson_lock__channel_xxx (xx represents the name of the lock) channel, listen for messages using asynchronous threads, and then use Semaphore in Java to block the current thread.

Once another client unlocks it, redission sends an unlock message to the redisson_lock__channel_xxx.

When the asynchronous thread receives the message, Semaphore will be called to release the semaphore, thus waking up the currently blocked thread to lock it.

Ps: this is only a brief description of the principle of redission locking. For the sake of space, there is no longer message parsing source code.

Interested partners can take a look at the source code of the redission lock.

Through the publish and subscribe mechanism, blocked threads can be awakened in time, reduce invalid idle queries, and effectively improve the efficiency of locking.

Ps: in this way, the performance is indeed improved, but the complexity of implementation is also very high. There is something about this part of the source code.

At this point, the study on "how to implement Redis publish and subscribe" is over. I hope to be able to solve your doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

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.