背景

在各个 IT 行业的公司,我们会有大大小小的业务需求。当每个产品的业务功能越来越繁重时,也许用户的需求其实很简单,就想 One Click。但是,其实这一个按钮背后可能有很多的系统交互的操作在进行,这就涉及到业务数据操作的事务,涉及到每个系统的交互逻辑、先后顺序以及数据的一致性。这些都需要在设计的时候,需要考虑到的问题。

浅谈解耦合

业务系统的设计有多重要 在 今天被问微服务,这几点,让面试官刮目相看 一文中,我们讲过微服务设计时的方方面面。其中核心的六个字可能就是:“高内聚,低耦合”。高内聚,我们在那篇文章中已经讲的很清楚了。那么低耦合,这就涉及到我们业务系统的设计了。所谓低耦合,就是要每个业务模块之间的关系降低,减少冗余、重复、交叉的复杂度,模块功能划分也尽可能单一。这样,才能达到低耦合的目的。
在电商行业,主要的功能就是购物,至于其他的,都是为购物作铺垫、营销手段:直播、促销、发优惠券等。从用户的角度来说,其实网上 shopping 的逻辑很简单:选中想要买的,支付 money 就 OK 了。但对于网站,或者说运营服务平台来说,其逻辑远没有那么简单。下面是一个简单的购物流程图:
在这里,我们看到,就这个简单的购物流程,对于用户来说,可能操作很简单:打开网站,登录后选择商品和选中收货地址支付,坐等收货。对于平台,其实它也不简单,包括了很多系统:用户系统、商品系统、仓库系统、订单系统、支付系统、物流系统等等。
不能仅仅因为客户的需求,只是下了一个订单,买了一件商品,那系统就设计一个就认为能解决所有事情,这种认识,可能一开始就是错的。这样的业务设计后,不但导致业务系统的逻辑很笨重,也会导致代码的 code review 非常之复杂。我曾经就亲自目睹过:好几个事情都是一个代码块来处理,甚至都写到几千行,甚至上万行。这样的思路,虽然可以实现暂时的需求。但是从长远角度,这是一个很要命的事情:这样的设计不仅仅说 code review 很吃力,兼容新功能也是很麻烦的,让后来者无法下手。而且长期下去,会导致表的死锁,甚至进入系统瘫痪状态。

如何解耦合

业务的复杂性,其实根本原因是没有把其给拆解化。如果把整个的大业务拆解成若干个小的需求,那对于实现,就显得即一目了然,又能完美兼容其他任何问题。咱们还是拿购物说事,为什么每个购物 app 的系统设计都是这样的套路:选中商品后必须先加入购物车,选好地址信息,然后再统一去提交订单,最后才去支付 money 呢?难道系统直接简单点,选中后就支付不就解决了吗?那么网站何必搞得这么的麻烦,浪费时间、金钱,是为了折腾人?统统都不是。其实这也是网站开发最初想的事情,并不是说一件事情一口气能解决,就鲁莽的直接一口气解决。也许到时候,时间久了,人的精力没那么旺盛,变得虚弱的时候,那一口气就无法完成了。网站也是,一个需求也许可以简单的设计,就能完成。但是如果仅仅想着,现在简单的就完成,那是对以后的不负责任。以后可能会出现一些难以想象的事情,并且难以解决。
上面扯远了,回归到解耦合,解耦合其实有很多办法。比如 Java 中就有很多解决低耦合的方法:监听、观察模式、异步回调、定时任务、消息中间件等等。

1.1 监听

在Java 里,有很多设计模式:工厂模式、单例模式、建造者模式、代理模式、解释器模式、监听模式、观察者模式等等。其中,监听模式是低耦合解决的方案之一。
所谓监听模式:事件源经过事件的封装传给监听器,当事件源触发事件后,监听器接收到事件对象可以回调事件的方法。这其中涉及到三个信息:事件源、事件、监听器。
For example : 模拟某个服务启动后,发送通知信息。
事件源:

  1. package com.damon.event;
  2. import java.util.ArrayList;
  3. import java.util.List;
  4. public class Context {
  5. private static List<Listener> list=new ArrayList<Listener>();
  6. public static void addListener(Listener listener){
  7. list.add(listener);
  8. }
  9. public static void removeListener(Listener listener){
  10. list.remove(listener);
  11. }
  12. public static void sendMsg(Event event){
  13. for(Listener listener:list){
  14. listener.onChange(event);
  15. }
  16. }
  17. }

事件:

  1. package com.damon.event;
  2. public class Event {
  3. public static final int INSTALLED = 1;
  4. public static final int STARTED = 2;
  5. public static final int RESOLVED = 3;
  6. public static final int STOPPED = 4;
  7. private int type;
  8. private Object source;
  9. public Event(int type, Object source) {
  10. this.type = type;
  11. this.source = source;
  12. }
  13. public int getType() {
  14. return type;
  15. }
  16. public Object getSource() {
  17. return source;
  18. }
  19. }

监听器:

  1. package com.damon.event;
  2. public class MyListener implements Listener {
  3. @Override
  4. public void onChange(Event event) {
  5. switch(event.getType()){
  6. case Event.STARTED :
  7. System.out.println("started...");
  8. break;
  9. case Event.RESOLVED :
  10. System.out.println("resolved...");
  11. break;
  12. case Event.STOPPED :
  13. System.out.println("stopped...");
  14. break;
  15. default:
  16. throw new IllegalArgumentException();
  17. }
  18. }
  19. }

测试:

  1. package com.damon.event;
  2. public class EventTest {
  3. public static void main(String[] args) {
  4. Listener listener = new MyListener();
  5. //加入监听者
  6. Context.addListener(listener);
  7. //启动完毕事件触发
  8. Context.sendMsg(new Event(Event.STARTED, new MyBundle()));
  9. }
  10. }

在服务启动的操作中,我们不需要等待或者去处理,而是继续其他的逻辑,等到服务启动后,事件监听器监听后会进行相应的操作。这样,就不会在服务启动的过程中,需要等待其启动,因为其启动的时间是无法估量的。所以就很好的解决其耦合性的问题。避免用户在等待过程中,浪费了大量不应该由用户承担的时间成本。毕竟,对于用户来说,时间就是金钱。

1.2 观察者模式

观察者模式,听着跟上面讲的监听模式有点像。但是,还是有区别的。所谓观察者模式:观察者相当于事件监听者,被观察者相当于事件源和事件,执行逻辑时通知观察者即可触发其 update,同时可传被观察者和其参数。看着是不是像简化了事件监听机制的实现。其又可以叫发布-订阅模式,只有两个角色。
For example : 微信群里发布了一条公告:下午三点开会,有些在群里的人接收到了消息去开会,但是有些人未在群里,未收到公告,被领导主动喊去开会。
观察者:

  1. public abstract class Observer {
  2. protected String name;
  3. protected Subject subject;
  4. public Observer(String name, Subject subject) {
  5. this.name = name;
  6. this.subject = subject;
  7. }
  8. public abstract void update();
  9. }

通知者:

  1. public interface Subject {
  2. //增加
  3. public void attach(Observer observer);
  4. //删除
  5. public void detach(Observer observer);
  6. //通知
  7. public void notifyObservers();
  8. //状态
  9. public void setAction(String action);
  10. public String getAction();
  11. }

具体人:群管理员

  1. public class WechatManager implements Subject {
  2. //同事好友列表
  3. private List<Observer> observers = new LinkedList<>();
  4. private String action;
  5. //添加
  6. @Override
  7. public void attach(Observer observer) {
  8. observers.add(observer);
  9. }
  10. //删除
  11. @Override
  12. public void detach(Observer observer) {
  13. observers.remove(observer);
  14. }
  15. //通知
  16. @Override
  17. public void notifyObservers() {
  18. for(Observer observer : observers) {
  19. observer.update();
  20. }
  21. }
  22. //状态
  23. @Override
  24. public String getAction() {
  25. return action;
  26. }
  27. @Override
  28. public void setAction(String action) {
  29. this.action = action;
  30. }
  31. }

具体观察者:群内人员与群外人员

  1. public class InWechatRoomObserver extends Observer {
  2. public InWechatRoomObserver(String name, Subject subject) {
  3. super(name, subject);
  4. }
  5. @Override
  6. public void update() {
  7. System.out.println(subject.getAction() + "\n" + name + "收到公告,去开会了");
  8. }
  9. }
  1. public class Test {
  2. public static void main(String[] args) {
  3. //群管理员为通知者
  4. WechatManager ma = new WechatManager();
  5. InWechatRoomObserver in = new InWechatRoomObserver("tom", ma);
  6. OutWechatRoomObserver out = new OutWechatRoomObserver("damon", ma);
  7. //群管理员通知
  8. ma.attach(out);
  9. ma.attach(in);
  10. //damon没在群内,未被通知到,所以被领导发现
  11. ma.detach(out);
  12. //老板回来了
  13. ma.setAction("下午三点,大家在大会议室开会");
  14. //发通知
  15. ma.notifyObservers();
  16. }
  17. }

可以看到:当一个对象的改变需要同时改变其它对象,并且它不知道具体有多少对象有待改变的时候,可考虑使用观察者模式。
即使用观察者模式的动机在于:在保证相关业务数据的一致性,我们不希望为了维持一致性而使各个逻辑紧密耦合,这样会给维护、扩展和重用都带来不便,而观察者模式所做的工作就是在解除耦合。

1.3 异步

异步,对于一个系统来说,异步操作可以很好的解耦合,因为每一步操作不需要等待结果即可继续往下进行,不论中间操作是否成功。在 Java 中,常见的异步注解:@Async,解决相应如果需要很多操作,或者操作时耗时很长,而异步进行处理来解决相关问题。有时需要注解 @EnableAsync 配合,然后弄一个异步线程池,来进行线程异步调度管理。
异步线程池初始化 bean :

  1. package com.damon.task;
  2. import java.util.concurrent.Executor;
  3. import java.util.concurrent.ThreadPoolExecutor;
  4. import org.springframework.context.annotation.Bean;
  5. import org.springframework.context.annotation.Configuration;
  6. import org.springframework.scheduling.annotation.EnableAsync;
  7. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  8. /**
  9. * 异步任务执行bean
  10. *
  11. * @author Damon
  12. *
  13. */
  14. @EnableAsync
  15. @Configuration
  16. public class TaskPoolConfig {
  17. @Bean("taskExecutor")
  18. public Executor taskExecutor() {
  19. ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  20. executor.setCorePoolSize(10);
  21. executor.setMaxPoolSize(20);
  22. executor.setQueueCapacity(200);
  23. executor.setKeepAliveSeconds(60);
  24. executor.setThreadNamePrefix("taskExecutor-");
  25. executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  26. return executor;
  27. }
  28. }
  29. 异步调度方法类:
  30. package com.damon.task;
  31. import org.apache.commons.lang.StringUtils;
  32. import org.slf4j.Logger;
  33. import org.slf4j.LoggerFactory;
  34. import org.springframework.http.*;
  35. import org.springframework.scheduling.annotation.Async;
  36. import org.springframework.stereotype.Component;
  37. import java.math.BigDecimal;
  38. import java.text.DecimalFormat;
  39. import java.util.ArrayList;
  40. import java.util.List;
  41. import java.util.Random;
  42. /**
  43. *
  44. * 远程业务调用封装类
  45. *
  46. * @author Damon
  47. * @date 2019年3月19日 下午3:29:45
  48. *
  49. */
  50. @Component
  51. public class TaskService {
  52. private Logger logger = LoggerFactory.getLogger(getClass());
  53. public static Random random = new Random();
  54. /**
  55. * @description 异步任务计算耗时
  56. * @param start 开始时间
  57. * @param userId 用户id
  58. * @throws Exception
  59. */
  60. @Async("taskExecutor")
  61. public void doTaskOne(long start, String userId) throws Exception {
  62. logger.info(" 开始做任务一 to {}", start);
  63. Thread.sleep(random.nextInt(10000));
  64. long end = System.currentTimeMillis();
  65. logger.info("完成任务一,耗时:" + (end - start) + "毫秒");
  66. }
  67. }

异步可以常见于很多业务,比如异步发送短讯告诉用户,支付成功,异步发送日志到 ELK 系统等。

1.4 定时任务

对于定时任务,就是指制定系统的某个时刻或每隔一段时间去触发一些逻辑执行,这样来保证业务数据的一致性,消息的一致性,或者数据的实时性。
我们常在 Java 里用 @EnableScheduling 来引入定时器,然后定义一个异步定时调度 bean:

  1. package com.damon.task;
  2. import java.util.concurrent.Executor;
  3. import java.util.concurrent.ThreadPoolExecutor;
  4. import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
  5. import org.springframework.context.annotation.Bean;
  6. import org.springframework.context.annotation.Configuration;
  7. import org.springframework.scheduling.annotation.AsyncConfigurer;
  8. import org.springframework.scheduling.annotation.EnableAsync;
  9. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  10. /**
  11. *
  12. * 异步任务执行bean
  13. * @author Damon
  14. * @date 2019年7月17日 上午10:35:56
  15. *
  16. */
  17. @EnableAsync
  18. @Configuration
  19. public class TaskPoolConfig implements AsyncConfigurer {
  20. @Bean("asyncTask")
  21. public Executor taskExecutor() {
  22. ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  23. //线程池维护线程的最少数量
  24. executor.setCorePoolSize(10);
  25. //线程池维护线程的最大数量
  26. executor.setMaxPoolSize(20);
  27. // 缓存队列
  28. executor.setQueueCapacity(200);
  29. //允许的空闲时间
  30. executor.setKeepAliveSeconds(60);
  31. executor.setThreadNamePrefix("asyncTask-");
  32. //对拒绝task的处理策略
  33. executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
  34. return executor;
  35. }
  36. @Override
  37. public Executor getAsyncExecutor() {
  38. return null;
  39. }
  40. @Override
  41. public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
  42. return null;
  43. }
  44. }

其中,cron 从左到右(用空格隔开):
秒 分 小时 月份中的日期 月份 星期中的日期 年份 复制代码
上面的逻辑是每分钟去执行某个逻辑,这样的业务我们也可能存在,For example:股票系统中,建模等数据一般都是用 Oracle 来存储的,有时候业务可能是用 Mysql,这时,需要一个定时任务来跑数据,常见的叫 ETL,所以 ETL 的由来,就是这样来的。这样的操作肯定不能在发生业务操作时来进行,否则会因为业务数据的海量读取,导致 IO 的性能,甚至内存、CPU 都会飙升。再如统计某个业务场景的数据,都可以通过这种解耦合的方式来处理。

1.5 消息中间件

消息中间件的话,这个也是很多的,比如:redis、rocketmq、rabbitmq、zk等等。这些中间件技术都可以再一个复杂的业务流程起到至关重要得作用。
当我们需要做一个秒杀的功能时,可以用 redis 来作分布式锁,这个能起到缓冲系统压力的作用,同时可以做到秒杀锁。
当我们需要在处理一些业务逻辑时,需要告知其他方,这时候可以用 MQ 来作消息处理,防止处理流程的断续。
当我们需要发送一些消息给外部时,但又不希望耽误当前的业务处理,这时候,可以用 MQ 或 redis 来处理消息。

1.6 策略者模式

  1. // 策略模式
  2. // 定义一个策略类
  3. type IStrategy interface {
  4. do(int, int) int
  5. }
  6. // 策略实现:加
  7. type add struct{}
  8. func (*add) do(a, b int) int {
  9. return a + b
  10. }
  11. // 策略实现:减
  12. type reduce struct{}
  13. func (*reduce) do(a, b int) int {
  14. return a - b
  15. }
  16. // 具体策略的执行者
  17. type Operator struct {
  18. strategy IStrategy
  19. }
  20. // 设置策略
  21. func (operator *Operator) setStrategy(strategy IStrategy) {
  22. operator.strategy = strategy
  23. }
  24. // 调用策略中的方法
  25. func (operator *Operator) calculate(a, b int) int {
  26. return operator.strategy.do(a, b)
  27. }