小心 spring.jpa.open-in-view 導致 DB connection 被拿光

 spring.jpa.open-in-view

• spring boot 的 property, spring.jpa.open-in-view 預設是開啟的
• 開啟的話, OpenSessionInViewInterceptor 就會介入
• 收到 web request 的時候, 會開一個 Hibernate Session
• 如果用到 DB, 就會拿一個 DB Connection
• 完成 request, 就把 connection 關閉

目的

• 原本一個 entity 有像是 OneToMany 的關聯時, 預設都會 lazy.
• 需要特別啟用這個 collection, 就需要特別去初始化, 例如 Hibernate.initialize
• 如果沒有初始化就會遇到 LazyInitializationException

問題

• 如果處理一個 request 的時候會需要比較長的時間, 那 DB connection 就會被卡住
• BTW, 預設 DB connection 是 10條 ( https://github.com/brettwooldridge/HikariCP )
• 如果需要呼叫外部服務, 或是外部服務遇到 connection timeout 之類的問題, 會導致一個 request 花很多時間才結束
• 而導致 connection 很快被拿光, 而且無法拿到新的 connection 而爆炸..

怎麼辦?

disable open-in-view

• 把 spring.jpa.open-in-view, 把這個設成 false, 可以避免 connection 跟著 request
• 可是要注意如果有 lazy 的地方要 initialize 否則會遇到 lazy exception

限制 request 的時間

• 把時間可能拉長的設計跟 DB access 分開來
• 例如把對外部系統的呼叫分開來

Reference

更多OSIV細節: https://www.baeldung.com/spring-open-session-in-view

可能會需要調整 HikariCP 的 pool size, 參考: https://github.com/brettwooldridge/HikariCP

Spring Boot 的 open-in-view property: https://docs.spring.io/spring-boot/docs/current/reference/html/application-properties.html#application-properties.data.spring.jpa.open-in-view

Spring Cloud Stream Introduction - 1

Goal

描述如何套用 spring cloud stream + Kafka 以及概念.
適合只面對 Kafka, 不涵蓋進階議題

Example

https://github.com/axxdeveloper/study-practice/tree/spring-cloud-stream

Concepts

• 一個 application 的架構: 外部系統(middleware) -> input -> application process -> output -> 外部系統
• 在 Spring Cloud Stream
• 透過 Binder 來處理外部系統的細節. ex. spring-cloud-stream-binder-kafka
• 透過 Binding 來處理 input.

Example

• Setup
• Spring Boot Application

@SpringBootApplication

public class DemoApplication {

    public static void main(String[] args) {

        SpringApplication.run(DemoApplication.class, args);

    }

}

• Empty application.properties

Kafka topic consumer

• Consumer method (注意! 這裡的 method name: status 就是 Binding name, 會影響 application.properties)

@Bean

public Consumer<String> status() {

    return status -> System.out.println("Received " + status);

}

• 指定 input topic application.properties

# 如果會從多個 topic 拿 message 就需要在 spring.cloud.function.definition 這個 attribute 指定, 並且用分號隔開. 這裡是預先指定一個

spring.cloud.function.definition=status

# 可以看到這的 pattern: spring.cloud.stream.bindings.{bindingName}-in-0.destination, 用來指定 topic name

# in 的部分是說 input topic

# 0 則是這個 binding 的第一個 input

spring.cloud.stream.bindings.status-in-0.destination=status

• 再來就可以發訊息給 status 這個 topic

Function

透過 Function 可以處理 input -> process -> output

• Function: 收到一個訊息後, 在尾巴貼上 random suffix, 然後回傳

@Bean

public Function<String, String> randomNumberSuffix() {

    return val -> val + " => append suffix " + Math.random();

}

• 指定 input & output topic in application.properties

# 注意此時我們已經加上第二個 binding

spring.cloud.function.definition=status;randomNumberSuffix

# 指定 randomNumberSuffix 的 intput topic 是 randomNumberSuffix, output topic 則是 status, 也就是同一個 application 的另一個 binding

spring.cloud.stream.bindings.randomNumberSuffix-in-0.destination=randomNumberSuffix

spring.cloud.stream.bindings.randomNumberSuffix-out-0.destination=status

• 發訊息給 randomNumberSuffix topic

Supplier

Supplier 是會被系統自動 polling, 預設一秒 poll 一次, 也可以指定 cron, 要 3.2 版之後才支援 custom binding polling configuration.

• Supplier

@Bean

public Supplier<Date> mydate() {

    return () -> new Date();

}

• 指定 output topic 給 mydate 這個 Supplier binding

# 此時已經增加第三個 binding definition

spring.cloud.function.definition=status;randomNumberSuffix;mydate

# 指定 output topic 為 status, 就是同個 app 的 topic

spring.cloud.stream.bindings.mydate-out-0.destination=status

# 改變預設的 poll config 為 2 秒 poll 一次

spring.cloud.stream.poller.fixed-delay=2000

• Supplier 只要打開 app 就會自動被執行

Other Concepts

• Consumer Group
• 跟 Kafka Consumer Group 的概念一樣
• 同一個 group 裡面只會有一個 consumer 收到 message
• 不同 group 則都會收到訊息
• 預設每個 consumer 都是不同的 group (anomymous group)
• 透過 {binding}.group=xxx 來指定 groupName
• Durability
• 有指定 group, 則對 consumer 的 subscription 就會被保留, 即使這個 group 目前沒有 consumer, 等 consumer 回來, 就會接著收到訊息
• anonymous group 的 subscription 就不會被保留, 因此 anonymous group 容易收到 duplicated message
• Partition
• 一個 topic 可以被切成多個 partition, 每個 partition 會由固定的一個 consumer 接收資料

 

Spring Integration - Basic Terms

Reference

Spring Integration Basic Terms
後篇: https://www.isaacnote.com/2018/10/spring-integration-channels.html 

Terms

org.springframework.integration.Message<T>

用來封裝訊息, 成員包含 MessageHeader 與 Payload

org.springframework.integration.MessageHeaders

MessageHeaders 是 immutable 的, 包含一些預設的 attribute: id, timestamp, correlation id, and priority

Payloads

用來裝 message body, 可以自訂 transformer 來傳遞封包

Message Channel

Message Channel 用來傳遞封包, 也用來 decouple producer 與 consumer.

Message Channel 有兩種模式: 

1. point to point, 一個封包只會被一個 consumer 收到
2. public/subscribe: 一個封包會被多個 consumer 收到

Message Endpoints

endpoint 泛指 Spring Integration 裡面的各種 component.

• Message Adapter: 資料可從外部系統透過 adapter 送進 Spring Integration
• Transformer: 轉換訊息
• Filter: 決定 Message 是否傳給 Message Channel
• Router: 透過 Message 的內容判斷要送給哪一個 Message Channel
• Splitter: 把一個 Message 切成好幾個轉送給不同適合的 Message Channel
• Aggregator: 把多個訊息合併成一個
• Service activator: Message Channel 與 Service instance 之間的介面

Example

spring-context.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:int="http://www.springframework.org/schema/integration"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
        http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-5.0.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">
        <context:component-scan base-package="examples" />
        <int:channel id="input"/>
        <int:channel id="output">
            <int:queue capacity="10"/>
        </int:channel>
        <int:service-activator input-channel="input"
                           output-channel="output"
                           ref="messageHandler"/>
</beans>

examples/MessageHandler.java

@Component
public class MessageHandler {
    @ServiceActivator
    public String handleMessage(String message) {
        System.out.println("Received message: " + message);
        return "MESSAGE:" + message;
    }
}

examples/Application.java

public class Application {
    public static void main(String[] args) {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:spring-context.xml");
        context.start();

        MessageChannel input = context.getBean("input", MessageChannel.class );
        PollableChannel output = context.getBean("output", PollableChannel.class );

        Scanner scanner = new Scanner(System.in);
        while (scanner.hasNextLine()) {
            input.send(MessageBuilder.withPayload(scanner.nextLine()).build());
        }
    }
}

Run, input message and send to MessageHandler to print