架构模式是软件架构中在给定环境下常遇到问题的通用的、可重用的解决方案。类似于软件设计模式但覆盖范围更广，致力于软件工程中不同问题，如计算机硬件性能限制、高可用性、业务风险极小化。一些架构模式在软件框架被实现。- 维基百科

说明

架构模式有很多种，本文只讨论工作中使用较多的几种：

• 分层架构
• Pipeline架构
• 事件驱动架构

 

分层架构

分层架构模式

分层架构模式工作中用的比较多，常见的有MVC等，通过分层将职责划分到某一层上，层次清晰，架构明了。

我们以MVC来举例说明：controller -> service -> dao

@RestController
@RequestMApping("/order")
public class OrderController {
    @Autowired
    private OrderService orderService;

    /**
     * 新增订单
     * @param order
     * @return
     */
    @PostMapping("/add")
    public Response addOrder(Order order) {
        orderService.add(order);
        return Response.success();
    }
}

public interface OrderService {
    /**
     *  添加订单
     * @param order
     * @return
     */
    boolean add(Order order);
}

public interface OrderRepository {

    int save(Order order);
}

按照依赖方向，上层依次依赖下层，每一层处理不同到逻辑。

之前到文章有讨论过通过依赖反转来改变依赖关系，从而更少到减少耦合。

 

Pipeline架构

Pipeline架构模式

Pipeline架构也称为管道或流水线架构，处理流程成线性，各个环节有相应到组件处理，从前到后顺序执行。

概念说明：

source: 数据源，通常使用流数据为源，比如：KafkaSource；

channel：信道或管道，用于处理或转换数据，比如：JsonChannel；

Sink：数据落地，通常用于数据存储或转发，比如：DbSink, KafkaSink；

Component: 组件，用于执行逻辑的最小单元，source,channel,sink都是一个Component；

Pipeline: 管道或流水线，一个Pipeline由上面的组件组成，不同的业务可以组装成不同的Pipeline；

 

代码实现：数字数据源 -> 累加 -> 转成字符串 -> 落地

/**
 *  组件
 */
public interface Component<T> {
    /**
     *  组件名称
     * @return
     */
    String getName();

    /**
     *  获取下游组件
     * @return
     */
    Collection<Component> getDownStrems();

    /**
     *  组件执行
     */
    void execute(T o);
}

public abstract class AbstractComponent<T, R> implements Component<T>{

    @Override
    public void execute(T o) {
        // 当前组件执行
        R r = doExecute(o);
        System.out.println(getName() + " receive " + o + " return " + r);
        // 获取下游组件，并执行
        Collection<Component> downStreams = getDownStrems();
        if (!CollectionUtils.isEmpty(downStreams)) {
            downStreams.forEach(c -> c.execute(r));
        }
    }

    protected abstract R doExecute(T o);
}

/**
 *  数据来源
 */
public abstract class Source<T, R> extends AbstractComponent<T, R>{

}

/**
 *  管道/信道
 * @param <T>
 */
public abstract class Channel<T, R> extends AbstractComponent<T, R> {

}

/**
 *  数据落地
 * @param <T>
 */
public abstract class Sink<T, R> extends AbstractComponent<T, R> {

}

public class IntegerSource extends Source<Integer,  Integer>{

    @Override
    protected Integer doExecute(Integer o) {
        return o;
    }

    @Override
    public String getName() {
        return "Integer-Source";
    }

    @Override
    public Collection<Component> getDownStrems() {
        return Collections.singletonList(new IncrChannel());
    }

}

public class IncrChannel extends Channel<Integer, Integer> {

    @Override
    protected Integer doExecute(Integer o) {
        return o + 1;
    }

    @Override
    public String getName() {
        return "Incr-Channel";
    }

    @Override
    public Collection<Component> getDownStrems() {
        return Collections.singletonList(new StringChannel());
    }

}

public class StringChannel extends Channel<Integer, String> {

    @Override
    protected String doExecute(Integer o) {
        return "str" + o;
    }

    @Override
    public String getName() {
        return "String-Channel";
    }

    @Override
    public Collection<Component> getDownStrems() {
        return Collections.singletonList(new StringSink());
    }

}

public class StringSink extends Sink<String, Void>{

    @Override
    protected Void doExecute(String o) {
        return null;
    }

    @Override
    public String getName() {
        return "String-Sink";
    }

    @Override
    public Collection<Component> getDownStrems() {
        return null;
    }

}

/**
 *  流水线
 */
public class Pipeline {
    /**
     *  数据源
     */
    private Source source;

    public Pipeline(Source source) {
        this.source = source;
    }

    /**
     *  启动
     */
    public void start() {
        source.execute(1);
    }
}

测试：

public class PipelineTest {

    @Test
    public void test() {
        Pipeline pipeline = new Pipeline(new IntegerSource());
        pipeline.start();
    }
}

执行结果：

Integer-Source receive 1 return 1

Incr-Channel receive 1 return 2

String-Channel receive 2 return str2

String-Sink receive str2 return null

 

事件驱动架构

事件驱动模式

事件驱动是以某个具体事件为触发条件，从而贯穿这个处理流程。通常事件驱动属于发布订阅模式或观察者模式， 用于异步处理，解耦业务逻辑。具体实现有进程内的和分布式的方式，比如：EventBus, MQ等等。

 

代码举例：

public class OrderEventListener implements Listener<OrderEvent> {

    @Override
    public void onEvent(OrderEvent event) {
        System.out.println("receive event: " + event);
    }
}

public class EventBus {

    private final static List<Listener> listeners = new ArrayList<>();

    /**
     *  注册监听器
     * @param listener
     */
    public static void registerListener(Listener listener) {
        listeners.add(listener);
    }

    /**
     *  发布事件
     * @param event
     */
    public void publishEvent(Event event) {
        // 收到并处理事件
        listeners.forEach(l -> {
            l.onEvent(event);
        });
    }
}

测试：

public class EventBusTest {

    @Test
    public void publish() {
        OrderEvent event = new OrderEvent("order_2", OrderState.PENDING_PAYMENT);
        EventBus.registerListener(new OrderEventListener());
        EventBus eventBus = new EventBus();
        eventBus.publishEvent(event);
    }
}

Spring中也有事件发布和监听(深入浅出Spring/SpringBoot 事件监听机制)：

@Component
public class OrderEventListener  {

    @Async
    @EventListener(OrderEvent.class)
    public void onEvent(OrderEvent event) {
        System.out.println("receive event: " + event);
    }
}

public class EventTest {
    @Autowired
    private ApplicationContext context;

    @Test
    public void publishEvent() {
        OrderEvent event = new OrderEvent("order_1", OrderState.PENDING_PAYMENT);
        context.publishEvent(event);
    }
}

 

总结

以上通过代码实例简单说明了工作中常用到的架构模式，但是模式不是固定的，工作中需结合实际情况按需使用即可。