rabbitmq 单应用 分布式

松耦合

Posted by Sinsy on February 14, 2021 About 5 k words and Need 15 min

在这里，先祝大家在新的一年里变得更强~

什么是耦合

耦合性(Coupling)，也叫耦合度，是对模块间关联程度的度量。耦合的强弱取决于模块间接口的复杂性、调用模块的方式以及通过界面传送数据的多少。模块间的耦合度是指模块之间的依赖关系，包括控制关系、调用关系、数据传递关系。模块间联系越多，其耦合性越强，同时表明其独立性越差( 降低耦合性，可以提高其独立性)。软件设计中通常用耦合度和内聚度作为衡量模块独立程度的标准。划分模块的一个准则就是高内聚低耦合。

在单应用解耦

通常来说，对于整个系统不必要，但是又要记录的，重复度比较高的。

比如：

  • 日志
  • 邮件
  • 提示

这类功能，通常需要和实际功能打交道的。第一想法可能是 AOP 做一个切面编程。完成松耦合。

针对上诉问题，我通常使用三步来完成。

  1. 抽象
  2. 切面
  3. 异步

抽象

我首先会，抽象出方法名，使用模板模式，满足不同模块对不同消息的使用。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

public abstract class Msg {

    public abstract void log();

    public abstract void email();

    public abstract void tip();

    public static AfterReturningMsg newAfterReturningMsg() {
        return new AfterReturningMsg();
    }

}

public class AfterReturningMsg extends Msg {

    @Override
    public void log() {
        // 日志操作
    }

    @Override
    public void email() {
        // 邮件操作
    }

    @Override
    public void tip() {
        // 提示操作
    }

}

这里有个更好的，就是在 Msg 的这个类中采用工厂 + 模板，这样松耦合以及扩展都不错。 之后扩展的类都继承 Msg 这个封装类，并实现子类，向上转型。例子请看 AfterReturningMsg 类。

切面

由于邮件、日志等功能，只需要等待业务正确，就可以操作的功能，我们使用 @AfterReturning 注解，通过切 controller 层，来完成方法增强。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

@Aspect
@Component
public class MsgAop {

    @Pointcut("execution(public * com.example.demo.controller.*.*(..))")
    public void pointCut(){}

    @AfterReturning(pointcut = "pointCut()")
    public void sendMsg() {
        Msg msg = Msg.newAfterReturningMsg();

        msg.email();
        msg.log();
        msg.tip();
    }

}

异步

使用 springBoot 的 @Async 注解，fork 一个子线程，帮助我们完成业务功能。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

@EnableAsync
@Aspect
@Component
public class MsgAop {

    @Pointcut("execution(public * com.example.demo.controller.*.*(..))")
    public void pointCut(){}

    @Async
    @AfterReturning(pointcut = "pointCut()")
    public void sendMsg() {
        Msg msg = Msg.newAfterReturningMsg();

        msg.email();
        msg.log();
        msg.tip();
    }

}

这里有个坑，HttpServletRequest 类，如果是在异步处理了，spring boot 会提前结束 servelet 生命周期，使得在使用这个类的时候，容易出现空指针异常。不过也挺好解决的。既然知道它会提前结束了。那么就复制这个类的所有传递下去，或者改下线程池，使用异步回调。

至此，一个单应用解耦的大体框架就实现了。看起来挺好的，不过实际中还是有挺多小细节或者缺点，需要注意的。下面我们总结下使用这个体系的缺点。

缺点

  1. 使用到了 @Async 注解，意味着要关注线程池的里面的参数，要考虑到实际机器的物理性能
  2. 一个 Msg 类的模板，无法适用于全部业务，可能每个对于 Msg 里的方法参数要求不同

在分布式解耦

在分布式系统底下，可能存在多个数据源或者利用多台机器横向扩张整个系统的性能，如果还是用上面的写法，无论 AOP 的时候进行动态数据源的切换，会将代码和一些业务耦合在一起，不利于后续的维护，又或者是当单台机器来到瓶颈，出现单机器性能拖垮整个系统性能。

这里我采用的 rabbitMq 进行解耦 AOP 代码。

依然是三步走。

  • 抽象
  • 切面
  • 消费消息

抽象

抽象和在单体一样，这里使用模板 + 工厂，抽出公共的方法，同时将对象转成 JSON 字符串。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

public abstract class Msg {

    public abstract String log(Log log);

    public abstract String email(Email email);

    public abstract String tip(Tip tip);

    public static Msg factory(String className) {
        // 简单工厂
        if ("AfterReturningMsg".equals(className)) {
            return new AfterReturningMsg();
        }
        return null;
    }

}

public class AfterReturningMsg extends Msg{
    @Override
    public String log(Log log) {
        // 业务处理
        // 转JSON 字符串
        return JSONObject.toJSONString(log);
    }

    @Override
    public String email(Email email) {
        // 业务处理
        // 转JSON 字符串
        return JSONObject.toJSONString(email);

    }

    @Override
    public String tip(Tip tip) {
        // 业务处理
        // 转JSON 字符串
        return JSONObject.toJSONString(tip);

    }
}

首先会对进来的消息进行处理，处理的消息直接输出成 JSON 字符串

切面

在切面的时候，采用 rabbitmq 的 direct 模式，放进 rabbitmq。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

@EnableAsync
@Aspect
@Component
public class MsgAop {

    @Pointcut("execution(public * com.example.demo.controller.*.*(..))")
    public void pointCut(){}

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Async
    @AfterReturning(pointcut = "pointCut()")
    public void sendMsg() {

        Msg msg = Msg.factory("AfterReturningMsg");

        Email email = new Email();
        Log log = new Log();
        Tip tip = new Tip();
        // 业务处理
        rabbitTemplate.convertAndSend("directs", "email", msg.email(email));
        rabbitTemplate.convertAndSend("directs", "log", msg.log(log));
        rabbitTemplate.convertAndSend("directs", "tip", msg.tip(tip));

    }

}

异步进行消息投递。比如你可以分发到不同的 MQ 上面，或者负载不同机器。

消费消息

通过监听 rabbitmq 的消息队列，直接进行业务处理。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

@Component
public class RouterConsumer {

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,
                    exchange = @Exchange(value = "directs"),
                    key = {"email"}
            )
    })
    public void email(String message) {
        // 业务处理
    }

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,
                    exchange = @Exchange(value = "directs"),
                    key = {"log"}
            )
    })
    public void log(String message) {
        // 业务处理
    }

    @RabbitListener(bindings = {
            @QueueBinding(
                    value = @Queue,
                    exchange = @Exchange(value = "directs"),
                    key = {"tip"}
            )
    })
    public void tip(String message) {
        // 业务处理
    }
    
}

由于是直接绑定的 MQ 的，可以有多个机器上面的应用去读取 MQ，从而避免单机器瓶颈的问题，也利于扩展，只需要扩充下绑定的消息队列，匹配下 routingkey。

缺点

  1. 增加了整个系统复杂程度
  2. 虽然提升了整个系统吞吐量，但是对开发要求会更高一点

总结

实现千千万万，每个人都有自己见解，无论是在单应用、或者在分布式的场景底下，核心都是异步处理非核心业务，或者是用户不需要马上感知的功能，打造一个更容易扩展的系统功能，提升整个系统的吞吐量、以及性能。

除了以上我所列的日志、邮件、提醒等，其实在电商系统，比如扣库存、发订单这些等，也是用户不需要马上感知的，也是适用于上面的归纳总结，关键在于你是怎么看的。

声明

作者: Sinsy
本文链接：https://blog.sincehub.cn/2021/02/14/coupling/
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文声明。
如您有任何商业合作或者授权方面的协商，请给我留言：550569627@qq.com

引用

CATALOG
    FEATURED TAGS
    CI/CD 源码 JDK shell
    FRIENDS