spring:
  datasource:
    type: com.alibaba.druid.pool.DruidDataSource
    driverClassName: com.mysql.jdbc.Driver
    master: #主庫A
      type: com.alibaba.druid.pool.DruidDataSource
      driverClassName: com.mysql.jdbc.Driver
      url: jdbc:mysql://127.0.0.1:3306/A?useUnicode=true&characterEncoding=UTF-8&useOldAliasMetadataBehavior=true&useSSL=false&serverTimezone=GMT%2B8
      username: root
      password: 123456
    slave: #從庫B
      type: com.alibaba.druid.pool.DruidDataSource
      driverClassName: com.mysql.jdbc.Driver
      url: jdbc:mysql://192.168.1.12:3306/B?useUnicode=true&characterEncoding=UTF-8&useOldAliasMetadataBehavior=true&useSSL=true&serverTimezone=GMT%2B8
      username: root
      password: 123456

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.spring.boot.autoconfigure.DruidDataSourceBuilder;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.jdbc.datasource.LazyConnectionDataSourceProxy;
 
import javax.sql.DataSource;
import java.util.HashMap;
 
// 主從數(shù)據(jù)源配置
@Configuration
public class DataSourceConfiguration {
 
    DataSourceProperties masterDataSource = new DataSourceProperties();
    DataSourceProperties slaveDataSource = new DataSourceProperties();
 
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.master")
    public DataSource masterDataSource() {
        DruidDataSource druidDataSource = masterDataSource.setDataSource(DruidDataSourceBuilder.create().build());
        return druidDataSource;
    }
 
    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.slave")
    public DataSource slaveDataSource() {
        DruidDataSource druidDataSource = slaveDataSource.setDataSource(DruidDataSourceBuilder.create().build());
        return druidDataSource;
    }
 
    @Bean
    public DataSource routeDataSource() {
        RoutingDataSource routingDataSource = new RoutingDataSource() {{
            setDefaultTargetDataSource(masterDataSource());
            setTargetDataSources(new HashMap<Object, Object>() {{
                put(DbType.MASTER, masterDataSource());
                put(DbType.SLAVE, slaveDataSource());
            }});
        }};
 
        return routingDataSource;
    }
 
    @Bean
    @Primary
    public LazyConnectionDataSourceProxy lazyConnectionDataSourceProxy() {
        return new LazyConnectionDataSourceProxy(routeDataSource());
    }
 
}

import com.alibaba.druid.pool.DruidDataSource;
import lombok.Setter;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
 
/**
 * 數(shù)據(jù)源配置文件
 */
@Setter
@Configuration
@ConfigurationProperties(prefix = "spring.datasource.druid")
public class DataSourceProperties {
 
    private int initialSize = 10;
 
    private int minIdle = 10;
 
    private int maxActive = 50;
 
    private int maxWait;
 
    private int timeBetweenEvictionRunsMillis = 300000;
 
    private int minEvictableIdleTimeMillis = 60000;
 
    private int maxEvictableIdleTimeMillis = 7200000;
 
    private String validationQuery = "SELECT 1 FROM DUAL";
 
    private boolean testWhileIdle = true;
 
    private boolean testOnBorrow = true;
 
    private boolean testOnReturn = true;
 
    public DruidDataSource setDataSource(DruidDataSource datasource) {
        /** 配置初始化大小、最小、最大 */
        datasource.setInitialSize(initialSize);  //優(yōu)先級：application的spring.datasource.master.initialSize > application的spring.datasource.druid.initialSize > datasource.setInitialSize(20)和datasource.setInitialSize(initialSize)
        datasource.setMaxActive(maxActive);
        datasource.setMinIdle(minIdle);
        /** 配置獲取連接等待超時的時間 */
        // datasource.setMaxWait(maxWait);
        /** 配置間隔多久才進(jìn)行一次檢測，檢測需要關(guān)閉的空閑連接，單位是毫秒 */
        datasource.setTimeBetweenEvictionRunsMillis(timeBetweenEvictionRunsMillis);
        /** 配置一個連接在池中最小、最大生存的時間，單位是毫秒 */
        datasource.setMinEvictableIdleTimeMillis(minEvictableIdleTimeMillis);
        datasource.setMaxEvictableIdleTimeMillis(maxEvictableIdleTimeMillis);
        /**
         * 用來檢測連接是否有效的sql，要求是一個查詢語句，常用select 'x'。如果validationQuery為null，testOnBorrow、testOnReturn、testWhileIdle都不會起作用。
         */
        datasource.setValidationQuery(validationQuery);
        /** 建議配置為true，不影響性能，并且保證安全性。申請連接的時候檢測，如果空閑時間大于timeBetweenEvictionRunsMillis，執(zhí)行validationQuery檢測連接是否有效。 */
        datasource.setTestWhileIdle(testWhileIdle);
        /** 申請連接時執(zhí)行validationQuery檢測連接是否有效，做了這個配置會降低性能。 */
        datasource.setTestOnBorrow(testOnBorrow);
        /** 歸還連接時執(zhí)行validationQuery檢測連接是否有效，做了這個配置會降低性能。 */
        datasource.setTestOnReturn(testOnReturn);
        
        return datasource;
    }
 
}

import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.springframework.stereotype.Component;
 
// aop環(huán)繞增強(qiáng)切換數(shù)據(jù)源
@Slf4j
@Aspect
@Component
public class DBConnectionAOP {
 
    @Around("@annotation(connectToDB)")
    public Object proceed(ProceedingJoinPoint pjp, ConnectToDB connectToDB) throws Throwable {
        try {
            if (connectToDB.value().equals("MASTER")) {
                log.info("Master DB 配置");
                DBContextHolder.setDbType(DbType.MASTER);
            } else if (connectToDB.value().equals("SLAVE")) {
                log.info("Slave DB 配置");
                DBContextHolder.setDbType(DbType.SLAVE);
            } else {
                log.info("默認(rèn) DB 配置");
            }
 
            Object result = pjp.proceed();
            DBContextHolder.clearDbType();
 
            return result;
        } finally {
            DBContextHolder.clearDbType();
        }
    }
}

public class DBContextHolder {
 
    private static final ThreadLocal<DbType> contextHolder = new ThreadLocal<DbType>();
 
    public static void setDbType(DbType dbType) {
        if (dbType == null) {
            throw new NullPointerException();
        }
        contextHolder.set(dbType);
    }
 
    public static DbType getDbType() {
        return (DbType) contextHolder.get();
    }
 
    public static void clearDbType() {
        contextHolder.remove();
    }
 
}

import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
 
// 動態(tài)數(shù)據(jù)源路由配置
public class RoutingDataSource extends AbstractRoutingDataSource {
 
    // 決定使用哪個數(shù)據(jù)源
    @Override
    protected Object determineCurrentLookupKey() {
        return DBContextHolder.getDbType();
    }
 
}

// 數(shù)據(jù)源枚舉
public enum DbType {
    MASTER,
    SLAVE
}

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
 
// 數(shù)據(jù)源選擇注解
@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface ConnectToDB {
    // 默認(rèn)主數(shù)據(jù)庫
    String value() default "primary";
 
}

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;
 
/**
 * 異步線程池的配置類
 */
@Configuration
@EnableAsync
public class ExecutorConfig {
    private static final Logger logger = LoggerFactory.getLogger(ExecutorConfig.class);
 
    @Bean
    public Executor asyncServiceExecutor() {
        logger.info("注冊asyncServiceExecutor");
 
//        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        ThreadPoolTaskExecutor executor = new VisibleThreadPoolTaskExecutor();
        //配置核心線程數(shù)
        executor.setCorePoolSize(5);
        //配置最大線程數(shù)
        executor.setMaxPoolSize(50);
        //配置隊列大小
        // Set the capacity for the ThreadPoolExecutor's BlockingQueue. Default is Integer.MAX_VALUE.
        // Any positive value will lead to a LinkedBlockingQueue instance; any other value will lead to a SynchronousQueue instance.
        executor.setQueueCapacity(100);
        // 設(shè)置允許的空閑時間（秒）
        executor.setKeepAliveSeconds(60);
        //配置線程池中的線程的名稱前綴
        executor.setThreadNamePrefix("async-service-");
        // rejection-policy：當(dāng)pool已經(jīng)達(dá)到max size的時候，如何處理新任務(wù)；CALLER_RUNS：不在新線程中執(zhí)行任務(wù)，而是有調(diào)用者所在的線程來執(zhí)行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());
        //執(zhí)行初始化
        executor.initialize();
 
        return executor;
    }
 
 
}

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.concurrent.ListenableFuture;
 
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
 
/**
 * ThreadPoolTaskExecutor的子類，在父類的基礎(chǔ)上加入了日志信息，查看線程池的信息
 */
public class VisibleThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {
    private static final Logger logger = LoggerFactory.getLogger(VisibleThreadPoolTaskExecutor.class);
 
    @Override
    public void execute(Runnable task) {
        showThreadPoolInfo("1. do execute");
        super.execute(task);
    }
 
    @Override
    public void execute(Runnable task, long startTimeout) {
        showThreadPoolInfo("2. do execute");
        super.execute(task, startTimeout);
    }
 
    @Override
    public Future<?> submit(Runnable task) {
        showThreadPoolInfo("1. do submit");
        return super.submit(task);
    }
 
    @Override
    public <T> Future<T> submit(Callable<T> task) {
        showThreadPoolInfo("2. do submit");
        return super.submit(task);
    }
 
    @Override
    public ListenableFuture<?> submitListenable(Runnable task) {
        showThreadPoolInfo("1. do submitListenable");
        return super.submitListenable(task);
    }
 
    @Override
    public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
        showThreadPoolInfo("2. do submitListenable");
        return super.submitListenable(task);
    }
 
    private void showThreadPoolInfo(String prefix) {
        ThreadPoolExecutor executor = getThreadPoolExecutor();
 
        if (null == executor) {
            return;
        }
 
        logger.info("MaximumPoolSize：" + executor.getMaximumPoolSize() +
                "，CorePoolSize：" + executor.getCorePoolSize() +
                "，ThreadNamePrefix：" + this.getThreadNamePrefix() +
                "，prefix：" + prefix +
                "，TaskCount：" + executor.getTaskCount() +
                "，CompletedTaskCount：" + executor.getCompletedTaskCount() +
                "，ActiveCount：" + executor.getActiveCount() +
                "，PoolSize：" + executor.getPoolSize() +
                "，QueueSize：" + executor.getQueue().size()
        );
    }
 
}

@PostMapping("queryUserData.do")
    @VerifyLogin
    public RetInfo<Object> queryUserData(String temp, String channelId) throws JsonProcessingException {
        String mobile = UserUtil.getMobile();
        User user = userService.queryUserData(mobile, temp, channelId);
        return new RetInfo<>(RetEnum.SUCCESS.getCode(), RetEnum.SUCCESS.getMsg(), user);
    }

@Slf4j
@Service
@Transactional(rollbackFor = Exception.class)
public class UserServiceImpl implements UserService {
    @Autowired
    private UserLogService userLogService;
 
    @Override
    public User queryUserData(String mobile, String temp, String channelId) throws JsonProcessingException {
        // 切換slave從數(shù)據(jù)源查詢
        Date loginTime = userLogService.validateLoginByToday(mobile); // 查詢當(dāng)天登錄時間
 
        Date date = new Date();
        Integer userStatus = 0;
        
        // 以下所有操作都切換回master主數(shù)據(jù)源查詢和更新
        User user = userMapper.selectUserByMobile(mobile);
        // 業(yè)務(wù)代碼.....
        userMapper.insert(user);
}
}

    @PostMapping("queryUserData.do")
    public RetInfo<Object> queryUserData(String temp, String channelId) throws JsonProcessingException {
        long start = System.currentTimeMillis();
        String mobile = UserUtil.getMobile();
        HashMap<String, Object> map = new HashMap<>();
        // 切換slave從數(shù)據(jù)源查詢
        Date loginTime = userLogService.validateLoginByToday(mobile); // 查詢用戶當(dāng)天登錄時間
        // 切換回master主數(shù)據(jù)源查詢和更新
        User user = userService.queryUserData(mobile, loginTime, temp, channelId);
 
        return new RetInfo<>(RetEnum.SUCCESS.getCode(), RetEnum.SUCCESS.getMsg(), user);
    }

@Service
public class UserLogServiceImpl implements UserLogService {
 
    private UserBehaviorService userBehaviorService;
 
    public UserLogServiceImpl(UserBehaviorService userBehaviorService) {
        this.userBehaviorService = userBehaviorService;
    }
 
    @Override
    public Date validateLoginByToday(String mobile) {
        return userBehaviorService.validateLoginByToday(mobile);
    }
}

@Slf4j
@Service
public class UserBehaviorServiceImpl implements UserBehaviorService {
 
    private UserBehaviorMapper userBehaviorMapper;
 
    public UserBehaviorServiceImpl(UserBehaviorMapper userBehaviorMapper) {
        this.userBehaviorMapper = userBehaviorMapper;
    }
 
    @Transactional
    @ConnectToDB(value = "SLAVE") // 這就是切換數(shù)據(jù)源最重要注解
    @Override
    public Date validateLoginByToday(String mobile) {
            return userBehaviorMapper.validateLoginByToday(mobile, DateUtil.dateToString(new Date(), "yyyyMMdd"));
    }
}

@Mapper
public interface UserBehaviorMapper {
    /**
     * 查詢用戶當(dāng)天是否活躍
     * @param mobile
     * @param dateStr
     * @return Integer
     */
    Integer validateActive(@Param("mobile") String mobile, @Param("dateStr") String dateStr);
}