RedisTemplate實作訊息佇列并且批量插入資料。-有解無憂

早期由于生產環境業務量小，所以日志是一條一條commit的，運行也沒出過問題，

后來隨著業務擴大并發量上來后，日志寫入因為頻繁與資料庫打交道導致資料庫連接池經常占滿，直至程式崩潰，

因為日志并非需要實時回應，所以考慮改用異步+批量提交的方式，

為了緩解jvm記憶體壓力，采用redis做訊息佇列（因為原專案有集成過redis，公司不想使用其他mq增加維護成本），

所以在網上找了篇springboot整合redistemplate做訊息佇列的資料，稍微改了一下，

參考資料：https://blog.csdn.net/qq_38553333/article/details/82833273

首先是redisConfig，

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;


@Configuration
@EnableCaching //開啟注解
public class RedisConfig {
    /**
     * retemplate相關配置
     * @param factory
     * @return
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {

        RedisTemplate<String, Object> template = new RedisTemplate<>();
        // 配置連接工廠
        template.setConnectionFactory(factory);

        //使用Jackson2JsonRedisSerializer來序列化和反序列化redis的value值（默認使用JDK的序列化方式）
        Jackson2JsonRedisSerializer jacksonSeial = new Jackson2JsonRedisSerializer(Object.class);

        ObjectMapper om = new ObjectMapper();
        // 指定要序列化的域，field,get和set,以及修飾符范圍，ANY是都有包括private和public
        om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        // 指定序列化輸入的型別，類必須是非final修飾的，final修飾的類，比如String,Integer等會跑出例外
        om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        jacksonSeial.setObjectMapper(om);

        // 值采用json序列化
        template.setValueSerializer(jacksonSeial);
        //使用StringRedisSerializer來序列化和反序列化redis的key值
        template.setKeySerializer(new StringRedisSerializer());

        // 設定hash key 和value序列化模式
        template.setHashKeySerializer(new StringRedisSerializer());
        template.setHashValueSerializer(jacksonSeial);
        template.afterPropertiesSet();

        return template;
    }


    @Bean
    public RedisMessageListenerContainer container(RedisConnectionFactory redisConnectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory);
        return container;
    }
}

訊息物體Message

import com.alibaba.fastjson.JSON;
import lombok.Data;

import java.util.UUID;

@Data
public class Message {
    private String id;
    private Integer retryCount;
    private String content;
    private Integer status;
    private String topic;

    public Message() {
    }

    public Message(String topic, Object object) {
        this.id = UUID.randomUUID().toString().replace("-", "");
        this.retryCount = 0;
        this.content = JSON.toJSONString(object);
        this.status = 0;
        this.topic = topic;
    }
}

Redis訂閱管理，采用觀察者模式，

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

@Component
public class TopicSubscriber {
    private final Map<String, Set<String>> subscriberMap = new HashMap();

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
　　// 觀察者模式實作消費者注冊，
    public Boolean addConsumer(String topic, String consumer) {
        Set<String> consumerList = subscriberMap.get(topic);

        if (consumerList == null) {
            consumerList = new HashSet<>();
        }
        Boolean b = consumerList.add(consumer);
        subscriberMap.put(topic, consumerList);
        return b;
    }

    public Boolean removeConsumer(String topic, String comsumer) {
        Set<String> consumerList = subscriberMap.get(topic);
        Boolean b = false;
        if (consumerList != null) {
            b = consumerList.remove(comsumer);
            subscriberMap.put(topic, consumerList);
        }
        return b;
    }
　　//訊息廣播
    public void broadcast(String topic, String id) {
        if (subscriberMap.get(topic) != null) {
            for (String consumer : subscriberMap.get(topic)) {
                String key = String.join("_", topic, consumer, id);
                if (!redisTemplate.hasKey("fail_" + key)) {
                    redisTemplate.opsForValue().set(key, id);
                    redisTemplate.opsForList().leftPush(topic + "_" + consumer, topic);
                }
            }
        }

    }
}

然后是Redis發布者

import com.alibaba.fastjson.JSON;
import com.redis.mq.subscriber.TopicSubscriber;
import io.netty.util.CharsetUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

@Component
public class RedisPublisher {

    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    @Autowired
    TopicSubscriber subscriber;

    @PostConstruct
    public void init() throws Exception {
        // todo test thread
        /*new Thread(() -> {
            int count = 0;
            try {
                Thread.sleep(3000l);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            while (count < 14) {
                try {
                    Thread.sleep(100l);
                    Generate generate = new Generate();
                    generate.setIdNo("" + count);
                    this.publish("GenerateLog", generate);
                    count++;
                } catch (Exception e) {
                }
            }
        }).start();*/
    }
　　
    public void publish(String topic, Object content) {　　//訊息發布到redis
        Message message = new Message(topic, content);
        subscriber.broadcast(topic, message.getId());
        redisTemplate.getConnectionFactory().getConnection().publish(
                topic.getBytes(CharsetUtil.UTF_8), JSON.toJSONString(message).getBytes()
        );
    }
}

Redis消費者，實作MessageListener的onMessage就可以，為了易于擴展，這里使用了泛型，

import com.alibaba.fastjson.JSON;
import com.cache.redis.mq.subscriber.TopicSubscriber;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.lang.reflect.ParameterizedType;
import java.util.concurrent.TimeUnit;

public abstract class RedisListener<T> implements MessageListener {

    @Autowired
    protected RedisTemplate<String, Object> redisTemplate;

    @Autowired
    protected RedisMessageListenerContainer messageListenerContainer;

    @Autowired
    protected TopicSubscriber subscriber;

    @Override
    public void onMessage(org.springframework.data.redis.connection.Message message, byte[] bytes) {
        String name = this.getClass().getSimpleName();
        String topic = new String(message.getChannel());
        String content = new String(message.getBody());
        Message m = JSON.parseObject(content, Message.class);
        String key = String.join("_", topic, name, m.getId());

        Object b = redisTemplate.opsForList().rightPop(topic + "_" + name);
        if (b != null && b.equals(m.getTopic())) {
            T t = JSON.parseObject(m.getContent(),
                    ((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0]);
            handler(t);　　// 處理redis訊息，
            // set data expire.使用redis的expire介面直接丟棄消費過的資料，
            redisTemplate.expire(key, 1, TimeUnit.NANOSECONDS);
        } else {
            // todo retry
            redisTemplate.opsForValue().set("fail_" + key, content);
        }
    }

    protected abstract void handler(T t);
}

到這里，基礎的redisMq就差不多了，下面涉及具體的業務及批量插入，

首先，加一個logHander介面，

public interface LogHandler {
    void process();
}

寫一個抽象類繼承RedisListener并且實作LogHander，這里用到了redis的put和poll阻塞佇列，

因為使用了mybatisplus又不想重新寫mybatis foreach批量查詢陳述句，所以這里偷懶直接用mybatis的sqlsession的單條預編譯，批量commit，

import com.cache.redis.mq.RedisListener;
import com.server.log.store.LogStore;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.session.ExecutorType;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.listener.ChannelTopic;

import javax.annotation.PostConstruct;
import java.lang.reflect.ParameterizedType;
import java.util.List;

@Slf4j
public abstract class AbstractLogHandler<T, M> extends RedisListener<T> implements LogHandler {

    @Autowired
    SqlSessionFactory factory;

    @PostConstruct
    public void addListener() {
        messageListenerContainer.addMessageListener(this, new ChannelTopic(getTopic()));
        subscriber.addConsumer(getTopic(), this.getClass().getSimpleName());
        process();
    }

    @Override
    protected void handler(T t) {
        getStore().put(t);　　//阻塞直到能新寫入，這里其實可以加個超時時間，避免一直阻塞，
    }

    protected abstract String getTopic();

    protected abstract LogStore<T> getStore();

    protected void commit(List<T> data) {
        if (data =https://www.cnblogs.com/braska/p/= null || data.isEmpty()) return;
        SqlSession session = factory.openSession(ExecutorType.BATCH);
        try {
            M mapper = session.getMapper(
                    (Class) (((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[1])
            );
            save(data, mapper);
            session.commit();
        } catch (Exception e) {
            log.error(String.format("topic %s 資料批量寫入失敗，{}", getTopic()), e);
            session.rollback();
        }finally {
            session.close();
        }

        data.forEach(o -> o = null);
        data.clear();
    }

    protected abstract void save(List<T> data, M m);
}

LogStore阻塞佇列

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Slf4j
public class LogStore<T> {

    private static final Integer QUEUE_CAPACITY = 10000;

    private BlockingQueue<T> logQueue;

    public LogStore() {
        this(QUEUE_CAPACITY);
    }

    public LogStore(int capacity) {
        this.logQueue = new LinkedBlockingQueue<>(capacity);
    }

    public void put(T t) {
        try {
            logQueue.put(t);
        } catch (InterruptedException e) {
            log.info("logStore put exception:{}", e);
        }
    }

    public T poll(long seconds) {
        try {
            return logQueue.poll(seconds, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            return null;
        }
    }
}

到這里，基礎的業務代碼就寫的差不多了，然后我們看下具體的業務處理類怎么寫，

比如我們的注冊日志，只要實作抽象類AbstraceLogHandler就可以了

import comcommon.constant.Constant;
import com.common.po.RegLog;
import com.dao.mapper.RegLogMapper;
import com.server.log.store.LogStore;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
public class RegisterLogHandler extends AbstractLogHandler<RegLog, RegLogMapper> {
    private final LogStore<RegLog> store = new LogStore<>();
    private String topic = Constant.TOPIC_REGISTER_LOG;
    // todo 可配置
    private final Integer batchSize = 300;
    private final Integer waitSeconds = 2;

    ExecutorService executor = Executors.newSingleThreadExecutor();

    @Override
    protected String getTopic() {
        return this.topic;
    }

    @Override
    protected LogStore<RegLog> getStore() {
        return this.store;
    }

    @Override
    public void process() {
        executor.execute(() -> { 　　　//開啟執行緒從redis中poll資料，
            List<RegLog> data = https://www.cnblogs.com/braska/p/new ArrayList<>(batchSize);
            while (true) {
                RegLog generate = this.store.poll(waitSeconds);
                if (generate != null) {
                    if (data.size() >= batchSize) {
                        commit(data);
                    }

                    data.add(generate);
                } else {　　//處理不足batchSize的尾巴資料，
                    if (data.size() > 0) {
                        commit(data);
                    }
                }
            }
        });
    }

    @Override
    protected void save(List data, RegLogMapper mapper) {
        data.forEach(o -> {
            if (o.getRegNo() == null) {
                String genNo = UUID.randomUUID().toString();
                o.setRegNo(genNo);
            }
            mapper.insert(o);　　//因為不想寫mybatis的foreach陳述句，所以這里直接用mybatisplus的insert單條陳述句，到這里sqlssesion并沒有commit.
        });
    }
}

呼叫：

@Autowired
    protected RedisPublisher publisher;


publisher.publish(Constant.TOPIC_REGISTER_LOG, log);

轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/186229.html

標籤：Java

上一篇：面向物件

下一篇：Python 十大語法