Series bài viết Offline Concurrency Control:

Part 1: Offline Concurrency Control - Phần 1: Hiện tượng

Part 2:  (Bài viết này) Offline Concurrency Control - Phần 2: Optimistic - "Lời xin lỗi"

Part 3: Offline Concurrency Control - Phần 3: Optimistic và Inconsistent Read

***

Tóm tắt phần 1

Phần 1, chúng ta đã bàn về một số vấn đề cơ bản của Concurrency, một số điểm quan trọng cần nhắc lại để tiện theo dõi bài viết này:

• Các vấn đề cơ bản của Concurrency bao gồm: Lost Update và Inconsistent Read.
• Có hai loại giao dịch:
  • Giao dịch hệ thống (System Transaction), thường là giao dịch tại database, có đặc điểm là ngắn, đảm bảo ACID.
  • Giao dịch nghiệp vụ (Business Transaction), thường kéo dài và không đảm bảo an toàn khi mở một long-live transaction trong database. Do đó, một giao dịch nghiệp vụ thường sẽ được chia thành nhiều giao dịch hệ thống.

Ở phần trước, chúng ta đã có được cái nhìn sơ bộ về Concurrency và một số cơ chế khắc phục. Nếu bạn còn nhớ thì Optimistic là một “lời xin lỗi” của ứng dụng. Tuấn sẽ nhận được lời xin lỗi đó như sau:

Điểm mấu chốt của Optimistic thể hiện ở chính lời xin lỗi của ứng dụng: Dữ liệu đã lỗi thời. Làm thế nào để biết được dữ liệu đó ở tình trạng như vậy. Ta sẽ đi tìm câu trả lời thông qua lý thuyết và demo. 

Nhắc lại ví dụ 1:

Tại một ứng dụng quản lý hệ thống kho hàng, Tuấn – một nhân viên quản lý và là một end user của ứng dụng, mở một giao diện quản lý sản phẩm để chỉnh sửa mô tả một sản phẩm từ “Áo thun nam” thành “Áo thun nam cao cấp”. 

Trong lúc Tuấn đang gõ chữ cho phần mô tả mới của mình thì Hùng – cũng là một quản lý và cũng là một end user, cũng làm như vậy. Anh mở thông tin sản phẩm ra và cập nhật số lượng tồn kho từ 50 xuống còn 45 vì anh vừa được thông báo sau khi hàng tồn kho được kiểm kê. Dù xuất phát sau Tuấn nhưng Hùng lại thực hiện rất nhanh, nhanh hơn cả Tuấn. Anh lưu thông tin sản phẩm vào cơ sở dữ liệu. Lúc này, Tuấn mới làm xong phần việc của mình, anh cũng bấm “Lưu & Đóng”. Cả hai cùng vui vẻ vì đã hoàn thành một task của mình trong ngày hôm đó.

Tồn kho đúng phải là 45 chứ không phải 50.

Sequence diagram cho ví dụ:

Trong sequence diagram trên:

• Vùng bao màu đỏ thể hiện các System Transaction tại database. Đây là những giao dịch ngắn đặc trưng.
• Business Transaction của Tuấn bên trái thể hiện ở hai giao dịch ngắn với database (lấy về thông tin sản phẩm với getProduct() và updateProduct()), xen giữa là chỉnh sửa dữ liệu trên màn hình.
• Business Transaction của Hùng, nằm bên phải, cũng tương tự.
• Hùng và Tuấn cùng đọc được thông tin sản phẩm giống nhau nhưng trên giao diện ứng dụng, họ chỉnh sửa thông tin khác nhau. Hùng cập nhật trước và thành công. Tuấn cập nhật sau, hệ thống phát hiện ra dữ liệu bị lỗi thời, do đó, thông báo lỗi cho Tuấn.
• Kết quả của quá trình này là Tuấn cần lấy lại thông tin mới nhất về sản phẩm trong database và làm lại từ đầu.

Ta thấy đúng điều đã bàn tới ở phần 1, Optimistic cho phép tính xử lý đồng thời của hệ thống cao hơn: Nếu tại thời điểm đó, có hàng ngàn người khác cũng lấy sản phẩm ra xem, họ vẫn đảm bảo thực hiện được điều đó. Chỉ là nếu họ muốn cập nhật thì chỉ người đầu tiên mới có thể hoàn thành.

Cơ chế hoạt động

Trong hình 2, có hai giao dịch hệ thống cập nhật dữ liệu là hai giao dịch cuối của mỗi người, T2 và H2, Optimistic đảm bảo rằng mỗi giao dịch này chỉ thành công khi dữ liệu nó định cập nhật không có sự thay đổi so với dữ liệu lần cuối cùng nó đọc lên. Để làm được điều này, dữ liệu cần phải có thông tin “version - phiên bản”:

Cứ mỗi lần dữ liệu được cập nhật thành công, version sẽ tăng lên 1:

Hình 4 bên trên cho thấy dữ liệu Data được cập nhật 3 lần, do đó, version của nó hiện tại của nó là 2 (với version xuất phát là 0). Sau thời điểm T3, Tuấn và Hùng sử dụng các system transaction T1 và H1 tương ứng (xem hình 2) để lấy dữ liệu về ứng dụng trong session riêng của họ:

Tuấn và Hùng chỉnh sửa dữ liệu, ứng dụng sẽ nâng version của mỗi người lên:

Do cả hai cùng lấy dữ liệu có { version = 2 } nên sau khi chỉnh sửa, version của họ sẽ tăng lên thành 3. Hùng cập nhật vào database trước khiến cho dữ liệu dưới database có { version = 3 }:

Giao dịch H2 thành công vì Data của Hùng có { version=2 } trùng với version hiện tại trong database. Chú ý rằng, giao dịch nghiệp vụ của cả hai người được chia nhỏ thành các giao dịch hệ thống như đã mô tả trước đây. Tại thời điểm Hùng thực hiện thành công giao dịch H2, Tuấn không hề mở một kết nối nào tới database cả, anh chỉ thuần túy cập nhật data trên ứng dụng. Giờ là thời điểm Tuấn cập nhật data của anh ấy xuống database:

Giao dịch T2 của Tuấn gặp thất bại do dữ liệu mà anh ấy lấy lên từ database { version = 2 } đã lỗi thời so với hiện tại { version = 3 }.

Data Versioning

Ở ví dụ trên, chúng ta sử dụng một cột “version” để đánh phiên bản của dữ liệu. Ngoài cách này, một cột đạng timestamp cũng có thể là một lựa chọn. Tuy nhiên, trong môi trường triển khai ứng dụng nhân bản, thông thường một service được nhân thành nhiều instance khác nhau cùng sử dụng một database, cần chú ý có thể system clock của chúng có thể không đồng bộ gây ra sự sai lệch ở mức độ nào đó.

Demo

Chúng ta đã nắm được nguyên lý hoạt động cơ bản của Optimistic Concurrency Lock pattern, giờ là lúc thử xem code thực tế sẽ thế nào. Chắc bạn cũng sẽ đoán ra, chúng ta sẽ có nhiều cách khác nhau để bàn luận, mỗi cách sẽ thích hợp với tình huống khác nhau trong ứng dụng của bạn.

Điểm mấu chốt đã phân tích: Thêm version vào dữ liệu. Ở đây, chúng ta sẽ sử dụng một version dưới dạng integer – cũng là một cách làm thông thường.

Bảng PRODUCTS trong database H2 như sau:

CREATE TABLE IF NOT EXISTS products (
                                    id BIGINT AUTO_INCREMENT PRIMARY KEY,
                                    name VARCHAR(255) NOT NULL,
description VARCHAR(255),
quantity INT NOT NULL,
version INT DEFAULT 0 NOT NULL,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
);

Tương ứng với nó là Product entity.

Triển khai trong Layered Architecture

Phiên bản đầu tiên triển khai trên một kiến trúc Layered. Và như chúng ta đã biết qua bài Hexagonal, tầng entity và repository gắn chặt với database, là bảng PRODUCTS trong ví dụ này. Entity có dạng:

@Entity(name = "products")
@Data
public class Product {
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;

    private String name;
    private String description;
    private Integer quantity;

    @Column(nullable = false)
    private Integer version;

    @Column(name = "updated_at", nullable = false)
    private java.time.LocalDateTime updatedAt;
}

Cột version đóng vai trò quan trọng trong cơ chế Optimistic mà chúng ta đã xem.

Giờ, quan trọng nhất là tầng service chứa nghiệp vụ ProductService. Giao dịch nghiệp vụ của ta bao gồm hai giao dịch hệ thống:

1. Tìm sản phẩm theo ID, thể hiện ở: getProductByID(Long id): Product
2. Cập nhật thông tin sản phẩm, ở: updateProductV1(Long id, String description, Integer quantity, Integer version): void.

Optimistic sẽ hoạt động ở giao dịch số (2) tại updateProductV1. Method này, tham số đầu vào là tất cả các thuộc tính có thể chỉnh sửa được như description, quantity và các thông tin cố định như id, cuối cùng là version. Chú ý rằng version này là version của dữ liệu được lấy lên từ database, tức là trước khi được chỉnh sửa.

Triển khai method updateProductV1:

Bước 1: Kiểm tra phiên bản

Lấy product từ database lên với ID rồi kiểm tra phiên bản xem có trùng với phiên bản mà mình đã sửa trên màn hình hay không. Chỗ này, bạn có thể sử dụng phương thức tìm kiếm sâu hơn như productRepository.findByIdAndVersion(id, version).

Bước 2: Thiết lập dữ liệu
Xác lập các giá trị từ form nhập liệu của người dùng đồng thời nâng phiên bản theo những gì mà Optimistic yêu cầu:

if (description != null) {
    product.setDescription(description);
}
if (quantity != null) {
    product.setQuantity(quantity);
}

product.setVersion(product.getVersion() + 1);
product.setUpdatedAt(java.time.LocalDateTime.now());

Bước 3: Cập nhật vào database

productRepository.save(product);

Method mặc định của JpaRepository do Spring cung cấp, cập nhật dữ liệu product theo ID.

Thiết lập giao dịch hệ thống:

Phương thức updateProduct được bao bọc trong một giao dịch với khai báo @Transactional:

@Transactional
public void updateProductV1(Long id, String description, Integer quantity, Integer version) {
    …
}

Nhận xét

Bạn có thể thấy ở code trên, có một kẽ hở nhỏ giữa bước 1 và bước 3: Tại thời điểm kiểm tra phiên bản, dữ liệu có thể hợp lệ nhưng khi cập nhật, dữ liệu đó có thể đã trở nên lỗi thời. Dù khoảng thời gian giữa hai bước này có thể rất nhỏ nhưng trong những hệ thống có tính Concurrency cao, chắc chắn lỗi này sẽ xảy ra. Bạn nên nhớ rằng

Trong hệ thống, có những thứ bạn thấy là lỗi nhưng rất khó có thể xảy ra thì chắc chắn chúng sẽ xảy ra.

Điều này dẫn chúng ta tới phiên bản tốt hơn, triệt tiêu được kẽ hở bằng cách kết hợp cả bước 1 và bước 3 trong cùng một thao tác.

Cải tiến

Product entity: 

Thuộc tính version để lưu version mới, còn version cũ được đưa vào thuộc tính mới oldVersion. Khi cập nhật dữ liệu entity, ta cần phải cập nhật các thông tin này để Optimistic sử dụng.

@Transient
private Integer oldVersion;

ProductRepository: 

Thêm method updateProductWithVersion(Product product) để kết hợp thao tác kiểm tra và cập nhật như trên đã đề cập:

@Modifying
@Query("UPDATE Product p SET p.description = :#{#product.description}, " +
       "p.quantity = :#{#product.quantity}, " +
       "p.version = :#{#product.version}, " +
       "p.updatedAt = :#{#product.updatedAt} " +
       "WHERE p.id = :#{#product.id} AND p.version = :#{#product.oldVersion}")
int updateProductWithVersion(@Param("product") Product product);

Tại sao lại cần return về int? Đây là số lượng bản ghi được cập nhật bởi sql UPDATE thỏa mãn mệnh đề WHERE với ID và version. Có thể xảy ra các khả năng sau:

• return { int = 0 }: không có bản ghi nào thỏa mãn WHERE, do đó, nếu ID tồn tại thì chứng tỏ phiên bản này, oldVersion, không còn tồn tại trong database nữa, nghĩa là có một ai đó đã cập nhật trước ta. 
• return { int > 0 } (cụ thể là 1): phiên bản  thỏa mãn oldVersion tồn tại trong database, UPDATE thành công.

Ta cần phải xử lý kết quả trả về của updateProductWithVersion ở ProductService.updateProductV2:

product.setOldVersion(product.getVersion());
product.setVersion(product.getVersion() + 1);
product.setUpdatedAt(java.time.LocalDateTime.now());

int updatedRows = productRepository.updateProductWithVersion(product);

if (updatedRows == 0) {
    throw new RuntimeException("Conflict detected! The product has been modified by another user.");
}

Tại updateProductV2, có một số thay đổi nhỏ với thiết lập các version và xử lý kết quả trả về từ Repository. Nếu số lượng bản ghi được cập nhật là 0, có xung đột và cần throw exception, ngược lại là thành công.

Github:

Link: Offline-Concurrency-Demo

Trong demo này:

• Code phần này thuộc module: layered-service
• Database sử dụng H2

Sau khi start service (file vn.softwaredesign.offlineconcurrency.layeredservice.LayeredServiceApplication):

• H2 console: http://localhost:8080/h2-console/login.do login như hình dưới.
• Swagger: http://localhost:8080/swagger-ui/index.html

 

Tổng kết

Như vậy, ở bài viết này, chúng ta đã xem về cơ chế hoạt động chính của pattern Offline Optimistic Lock cùng với cài đặt một ví dụ đơn giản. Tuy nhiên, vấn đề mà chúng ta đã bàn mới chỉ thể hiện được một khía cạnh của Concurrency: xử lý Lost Update.

Ở bài tiếp theo, chúng ta sẽ vẫn bàn về Offline Optimistic Lock pattern nhưng ở một số khía cạnh nâng cao hơn:

• Optimistic trong vấn đề Inconsistent Read.
• Phương án triển khai khác Layered Architecture ở bài viết này.