Shulou(Shulou.com)05/31 Report--

Today, the editor will share with you the relevant knowledge points about how to achieve responsive Redis interaction in Spring. The content is detailed and the logic is clear. I believe most people still know too much about this knowledge, so share this article for your reference. I hope you can get something after reading this article.

This article will simulate a user service and use Redis as the data storage server.

Involves two java bean, users and rights and interests

Public class User {private long id; private String name; / / label private String label; / / ship to address Longitude private Double deliveryAddressLon; / / ship to address Dimension private Double deliveryAddressLat; / / latest check-in date private String lastSigninDay; / / Credit private Integer score; / / Rights private List rights;.} public class Rights {private Long id; private Long userId Private String name;.} start

Introduce dependency

Org.springframework.boot spring-boot-starter-data-redis-reactive

Add Redis configuration

Spring.redis.host=192.168.56.102spring.redis.port=6379spring.redis.password=spring.redis.timeout=5000

SpringBoot start

@ SpringBootApplicationpublic class UserServiceReactive {public static void main (String [] args) {new SpringApplicationBuilder (UserServiceReactive.class) .web (WebApplicationType.REACTIVE) .run (args);}}

After the application starts, Spring automatically generates ReactiveRedisTemplate (its underlying framework is Lettuce).

ReactiveRedisTemplate uses similar to RedisTemplate, but it provides asynchronous, responsive Redis interaction.

Again, responsive programming is asynchronous, the thread is not blocked after the ReactiveRedisTemplate sends the Redis request, and the current thread can perform other tasks.

After the Redis response data is returned, ReactiveRedisTemplate dispatches the thread to process the response data.

Responsive programming can implement asynchronous calls and handle asynchronous results in an elegant way, which is its greatest significance.

Serialization

The default serialization used by ReactiveRedisTemplate is Jdk serialization, which we can configure as json serialization

@ Beanpublic RedisSerializationContext redisSerializationContext () {RedisSerializationContext.RedisSerializationContextBuilder builder = RedisSerializationContext.newSerializationContext (); builder.key (StringRedisSerializer.UTF_8); builder.value (RedisSerializer.json ()); builder.hashKey (StringRedisSerializer.UTF_8); builder.hashValue (StringRedisSerializer.UTF_8); return builder.build ();} @ Beanpublic ReactiveRedisTemplate reactiveRedisTemplate (ReactiveRedisConnectionFactory connectionFactory) {RedisSerializationContext serializationContext = redisSerializationContext (); ReactiveRedisTemplate reactiveRedisTemplate = new ReactiveRedisTemplate (connectionFactory,serializationContext); return reactiveRedisTemplate;}

The builder.hashValue method specifies how the Redis list values are serialized, and since the Redis list values of this article only hold strings, they are still set to StringRedisSerializer.UTF_8.

Basic data type

ReactiveRedisTemplate supports basic data types such as Redis strings, hashes, lists, collections, ordered collections, etc.

This article uses hashes to save user information, lists to save user rights and interests, and the use of other basic data types is not expanded in this article.

Public Mono save (User user) {ReactiveHashOperations opsForHash = redisTemplate.opsForHash (); Mono userRs = opsForHash.putAll ("user:" + user.getId (), beanToMap (user)); if (user.getRights ()! = null) {ReactiveListOperations opsForRights = redisTemplate.opsForList (); opsForRights.leftPushAll ("user:rights:" + user.getId (), user.getRights ()) .subscribe (l-> {logger.info ("add rights: {}", l) });} return userRs;}

The beanToMap method is responsible for converting the User class to map.

HyperLogLog

The Redis HyperLogLog structure can count the number of different elements in a collection.

Use HyperLogLog to count the number of users logged in every day

Public Mono login (User user) {ReactiveHyperLogLogOperations opsForHyperLogLog = redisTemplate.opsForHyperLogLog (); return opsForHyperLogLog.add ("user:login:number:" + LocalDateTime.now (). ToString (). Substring (0,10), user.getId ());} BitMap

Redis BitMap (bitmap) uses a bit bit to represent the value or state of an element. Because Bit is the smallest unit of computer storage, using it for storage will save a lot of space.

Use BitMap to record whether users have checked in this week

Public void addSignInFlag (long userId) {String key = "user:signIn:" + LocalDateTime.now (). GetDayOfYear () / 7 + (userId > > 16); redisTemplate.opsForValue (). SetBit (key, userId & 0xffff, true) .subscribe (b-> logger.info ("set: {}, result: {}", key, b);}

The high 48 bits of userId are used to divide users into different key, and the lower 16 bits are used as bitmap offset parameter offset.

The offset parameter must be greater than or equal to 0 and less than 2 ^ 32 (bit mapping is limited to 512 MB).

Geo

Redis Geo can store geolocation information and calculate geolocation.

Such as finding warehouse information within a given range

Public Flux getWarehouseInDist (User u, double dist) {ReactiveGeoOperations geo = redisTemplate.opsForGeo (); Circle circle = new Circle (new Point (u.getDeliveryAddressLon (), u.getDeliveryAddressLat ()), dist); RedisGeoCommands.GeoRadiusCommandArgs args = RedisGeoCommands.GeoRadiusCommandArgs.newGeoRadiusArgs (). IncludeDistance (). SortAscending (); return geo.radius ("warehouse:address", circle, args);}

Warehouse:address this collection needs to first save the warehouse location information.

The ReactiveGeoOperations#radius method can find elements in the collection that are geographically within a given range, and it also supports operations such as adding elements to the collection and calculating the geographic distance between two elements in the collection.

Lua

ReactiveRedisTemplate can also execute Lua scripts.

The following is done through the Lua script to complete the user check-in logic: if the user does not check in today, allow check-in, points plus 1, if the user has checked in today, then refuse to accept the operation.

Public Flux addScore (long userId) {DefaultRedisScript script = new DefaultRedisScript (); script.setScriptSource (new ResourceScriptSource (new ClassPathResource ("/ signin.lua")); List keys = new ArrayList (); keys.add (String.valueOf (userId)); keys.add (LocalDateTime.now (). ToString (). Substring (0,10); return redisTemplate.execute (script, keys);}

The signin.lua content is as follows

Local score=redis.call ('hget','user:'..KEYS [1],' score') local day=redis.call ('hget','user:'..KEYS [1],' lastSigninDay') if (day==KEYS [2]) then return '0'else redis.call (' hset','user:'..KEYS [1], 'score', score+1,'lastSigninDay',KEYS [2]) return' 1'endStream

Redis Stream is a new data type added in Redis version 5.0. This type can implement message queue, provide message persistence and active / standby replication, remember the access location of each client, and ensure that the message is not lost.

Redis draws lessons from the design of kafka. There can be multiple consumer groups in a Stream and multiple consumers in a consumer group.

If a consumer in a consumer group consumes a message in Stream, the message will not be consumed by other consumers in that consumer group. Of course, it can also be consumed by a consumer in another consumer group.

The following defines a Stream consumer who is responsible for processing received equity data

@ Componentpublic class RightsStreamConsumer implements ApplicationRunner, DisposableBean {private static final Logger logger = LoggerFactory.getLogger (RightsStreamConsumer.class); @ Autowired private RedisConnectionFactory redisConnectionFactory; private StreamMessageListenerContainer container; / / Stream queue private static final String STREAM_KEY = "stream:user:rights"; / / Consumer group private static final String STREAM_GROUP = "user-service"; / / Consumer private static final String STREAM_CONSUMER = "consumer-1" @ Autowired @ Qualifier ("reactiveRedisTemplate") private ReactiveRedisTemplate redisTemplate Public void run (ApplicationArguments args) throws Exception {StreamMessageListenerContainer.StreamMessageListenerContainerOptions options = StreamMessageListenerContainer.StreamMessageListenerContainerOptions.builder () .batchSize / / maximum number of count pulled in a batch. Executor (Executors.newSingleThreadExecutor ()) / / thread pool .pollTimeout (Duration.ZERO) / / blocking polling .targetType (Rights.class) / / Target type (type of message content) .build () / / create a message listening container container = StreamMessageListenerContainer.create (redisConnectionFactory, options) / / prepareStreamAndGroup looks for Stream information, and if it does not exist, create Stream prepareStreamAndGroup (redisTemplate.opsForStream (), STREAM_KEY, STREAM_GROUP) .subscribe (stream-> {/ / create a consumer for Stream And bind the processing classes container.receive (Consumer.from (STREAM_GROUP, STREAM_CONSUMER), StreamOffset.create (STREAM_KEY, ReadOffset.lastConsumed ()), new StreamMessageListener ()) Container.start ();} @ Override public void destroy () throws Exception {container.stop ();} / find Stream information. If it does not exist, create a Stream private Mono prepareStreamAndGroup (ReactiveStreamOperations ops, String stream, String group) {/ / info method to query Stream information. If the Stream does not exist, the underlying layer will report an error, and the onErrorResume method will be called. Return ops.info (stream) .onErrorResume (err-> {logger.warn ("query stream err: {}", err.getMessage ()); / / createGroup method to create Stream return ops.createGroup (stream, group) .flatMap (s-> ops.info (stream));}) } / / message processing object class StreamMessageListener implements StreamListener {public void onMessage (ObjectRecord message) {/ / processing messages RecordId id = message.getId (); Rights rights = message.getValue (); logger.info ("receive id: {}, rights: {}", id, rights) RedisTemplate.opsForList () .leftPush ("user:rights:" + rights.getUserId (), rights) .subscribe (l-> {logger.info ("add rights: {}", l);});}

Let's take a look at how to send a message

Public Mono addRights (Rights r) {String streamKey = "stream:user:rights"; / / stream key ObjectRecord record = ObjectRecord.create (streamKey, r); Mono mono = redisTemplate.opsForStream () .add (record); return mono;}

Create a message logging object, ObjectRecord, and send information records through ReactiveStreamOperations.

Sentinel 、 Cluster

ReactiveRedisTemplate also supports Redis Sentinel and Cluster cluster modes, and you only need to adjust the configuration.

The Sentinel configuration is as follows

Spring.redis.sentinel.master=mymasterspring.redis.sentinel.nodes=172.17.0.4:26379172.17.0.5:26379172.17.0.6:26379spring.redis.sentinel.password=

Spring.redis.sentinel.nodes is configured with the Sentinel node IP address and port, not the Redis instance node IP address and port.

The Cluster configuration is as follows

Spring.redis.cluster.nodes=172.17.0.2:6379172.17.0.3:6379172.17.0.4:6379172.17.0.5:6379172.17.0.6:6379172.17.0.7:6379spring.redis.lettuce.cluster.refresh.period=10000spring.redis.lettuce.cluster.refresh.adaptive=true

For example, if node2 in Redis Cluster is a slave node of node1, this information will be cached in Lettuce. When node1 goes down, Redis Cluster will upgrade node2 to master node. However, Lettuce does not automatically switch the request to node2 because its buffer is not refreshed.

When spring.redis.lettuce.cluster.refresh.adaptive configuration is enabled, Lettuce can regularly refresh the cache information of Redis Cluster cluster, dynamically change the node situation of the client, and complete the failover.

For the time being, no solution has been found for ReactiveRedisTemplate to implement pipeline, transactions.

These are all the contents of the article "how to achieve responsive Redis interaction in Spring". Thank you for reading! I believe you will gain a lot after reading this article. The editor will update different knowledge for you every day. If you want to learn more knowledge, please pay attention to the industry information channel.

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.