本文深入探讨如何在java中构建一个高效且线程安全的排队取号系统。针对多部门并发取号的需求，文章提出了一种基于`concurrenthashmap`的部门级同步方案，避免了全局锁带来的性能瓶颈，确保了同一部门内的顺序性，同时允许不同部门间并行操作，从而优化系统并发性能。文章还强调了数据库层面并发控制的重要性，以提供端到端的数据一致性。
引言：并发取号系统的挑战
在许多业务场景中，例如银行、医院或服务中心，都需要一个排队取号系统来管理客户的顺序。当多个用户（或线程）同时请求获取下一个号码时，系统必须确保每个号码只能被分配一次，避免重复分配或跳号。更复杂的场景是，系统可能服务于多个部门，而不同部门的取号流程可以并行进行，但同一部门内的取号操作必须是严格顺序的。直接对整个取号函数进行同步（例如使用synchronized关键字修饰方法）虽然能保证线程安全，但会严重限制系统的并发能力，导致不同部门之间的操作也相互阻塞，从而降低整体性能。
传统同步方案的局限性
考虑以下简化版的取号函数伪代码：
private DailyTurns callTurnLocal(int userId) {
try {
DailyTurns turn = null;
// 1. 从数据库获取下一个可用号码
turn = getNextTurnForUser(userId); // 假设这里也需要部门ID
if (turn != null) {
// 2. 更新号码状态为“已叫号”，并分配给用户
turn.setTurnStatusId(TURN_STATUS_CALLED);
turn.setEventDate(new Date());
turn.setUserId(userId);
// 3. 保存更新到数据库
turn = save(turn);
}
return turn;
} catch (Exception e) {
// 异常处理
return null;
}
}如果直接将callTurnLocal方法声明为synchronized，那么在任何时刻，只有一个线程能够进入该方法。这意味着即使来自不同部门的两个用户同时请求取号，它们也必须排队等待，这显然不符合“不同部门可并行”的需求。为了实现部门级的并发，我们需要一种更细粒度的同步机制。
基于ConcurrentHashMap的部门级锁策略
为了解决上述问题，我们可以利用ConcurrentHashMap来为每个部门维护一个独立的锁对象。当一个线程需要为某个部门取号时，它会尝试获取该部门对应的锁。这样，同一部门的请求会因为争抢同一个锁而串行执行，而不同部门的请求则可以同时获取各自的锁，从而实现并行处理。
立即学习“Java免费学习笔记（深入）”；
Alibabacloud Sdk Client Initialization For Java
下载
在 Java 中初始化和管理阿里云 SDK客户端。包括单例模式、线程安全、endpoint 与 region 配置、VPC 终端节点、同步与异步等。
核心思想：
- 创建一个ConcurrentHashMap<Integer, Object>，其中键是部门ID（departmentId），值是用于该部门同步的任意Object实例。
- 当需要为某个部门执行取号操作时，首先从ConcurrentHashMap中获取或创建一个与该部门ID关联的锁对象。
- 使用synchronized关键字配合这个部门特定的锁对象，来包裹取号的核心逻辑。
实现细节与示例代码
首先，定义一个简单的DailyTurns实体类和TurnStatus常量，用于模拟业务数据：
import java.util.Date;
import java.util.concurrent.ConcurrentHashMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
// 模拟排队号码实体
class DailyTurns {
private int turnId;
private int departmentId;
private int userId;
private int turnStatusId;
private Date eventDate;
public DailyTurns(int turnId, int departmentId) {
this.turnId = turnId;
this.departmentId = departmentId;
}
// Getters and Setters (省略部分代码，实际应用中需完整)
public int getTurnId() { return turnId; }
public void setTurnId(int turnId) { this.turnId = turnId; }
public int getDepartmentId() { return departmentId; }
public void setDepartmentId(int departmentId) { this.departmentId = departmentId; }
public int getUserId() { return userId; }
public void setUserId(int userId) { this.userId = userId; }
public int getTurnStatusId() { return turnStatusId; }
public void setTurnStatusId(int turnStatusId) { this.turnStatusId = turnStatusId; }
public Date getEventDate() { return eventDate; }
public void setEventDate(Date eventDate) { this.eventDate = eventDate; }
@Override
public String toString() {
return "DailyTurns{" +
"turnId=" + turnId +
", departmentId=" + departmentId +
", userId=" + userId +
", turnStatusId=" + turnStatusId +
", eventDate=" + eventDate +
'}';
}
}
// 模拟号码状态常量
class TurnStatus {
public static final int TURN_STATUS_PENDING = 1; // 待叫号
public static final int TURN_STATUS_CALLED = 2; // 已叫号
// ... 其他状态
}接下来是包含部门级同步逻辑的取号服务类：
public class TurnService {
private static final Logger logger = LoggerFactory.getLogger(TurnService.class);
// 使用ConcurrentHashMap存储每个部门的锁对象
// 键是部门ID，值是用于同步的任意Object实例
private final ConcurrentHashMap<Integer, Object> departmentLocks = new ConcurrentHashMap<>();
// 模拟从数据库获取下一个可用号码
// **重要提示：此方法内部仍需数据库层面的并发控制，详见下文**
private DailyTurns getNextTurnForDepartment(int departmentId) {
// 模拟数据库查询耗时
try {
Thread.sleep((long) (Math.random() * 50));
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
logger.warn("Thread interrupted while simulating DB latency.", e);
}
// 实际场景中，这里会查询数据库，获取该部门下一个未被叫号的号码
// 并可能需要使用 SELECT ... FOR UPDATE 等数据库锁机制来预留号码
// 这里简化为生成一个模拟的号码
int nextTurnId = generateNextDummyTurnId(departmentId); // 模拟生成一个号码
if (nextTurnId > 0) {
return new DailyTurns(nextTurnId, departmentId);
}
return null;
}
// 模拟生成下一个号码ID（非线程安全，仅用于演示应用层逻辑）
private static final ConcurrentHashMap<Integer, Integer> departmentTurnCounters = new ConcurrentHashMap<>();
private int generateNextDummyTurnId(int departmentId) {
return departmentTurnCounters.compute(departmentId, (k, v) -> (v == null ? 1 : v + 1));
}
// 模拟将更新后的号码信息保存到数据库
private DailyTurns save(DailyTurns turn) {
// 实际中会将更新后的DailyTurns对象保存到数据库
logger.info("保存叫号信息： " + turn);
return turn;
}
/**
* 为指定用户和部门获取下一个号码。
* 该方法实现了部门级的线程安全。
*
* @param userId 用户ID
* @param departmentId 部门ID
* @return 获取到的号码信息，如果失败则返回null
*/
public DailyTurns callTurnLocal(int userId, int departmentId) {
DailyTurns resultTurn = null;
// 获取或创建该部门的锁对象。
// computeIfAbsent 是线程安全的，确保每个 departmentId 只有一个锁对象被创建。
// 如果 departmentId 对应的锁不存在，则会创建一个新的 Object 实例作为锁。
Object departmentLock = departmentLocks.computeIfAbsent(departmentId, k -> new Object());
// 对该部门的锁对象进行同步
synchronized (departmentLock) {
try {
DailyTurns fetchedTurn = null;
// 1. 获取下一个号码。
// 这一步在应用层已经由 departmentLock 保证了同一部门的串行访问。
// 但仍需注意数据库层面的并发控制。
fetchedTurn = getNextTurnForDepartment(departmentId);
if (fetchedTurn != null) {
resultTurn = fetchedTurn;
// 2. 更新号码状态和用户信息
resultTurn.setTurnStatusId(TurnStatus.TURN_STATUS_CALLED);
resultTurn.setEventDate(new Date());
resultTurn.setUserId(userId);
// 3. 保存更新到数据库
resultTurn = save(resultTurn);
}
return resultTurn;
} catch (Exception e) {
logger.error("为部门 " + departmentId + " 取号时发生异常: " + e.getMessage(), e);
return null;
}
}
}
// 示例主方法，模拟多线程并发取号
public static void main(String[] args) throws InterruptedException {
TurnService service = new TurnService();
// 模拟15个用户，分别来自部门1、2、3
for (int i = 0; i < 15; i++) {
final int user = i + 1;
final int dept = (i % 3) + 1; // 部门ID在1, 2, 3之间循环
new Thread(() -> {
DailyTurns turn = service.callTurnLocal(user, dept);
if (turn != null) {
System.out.println("用户 " + user + " (部门 " + dept + ") 成功获取号码: " + turn.getTurnId());
} else {
System.out.println("用户 " + user + " (部门 " + dept + ") 未能获取号码。");
}
}).start();
}
// 等待所有线程完成，以便观察输出
Thread.sleep(2000);
}
}在上述代码中，departmentLocks.computeIfAbsent(departmentId, k -> new Object()) 是关键。它会原子性地检查ConcurrentHashMap中是否存在departmentId对应的锁对象。如果存在，则返回该对象；如果不存在，则创建一个新的Object实例并放入map中，然后返回。这样，对于同一个departmentId，所有线程都会获取到并同步在同一个Object实例上，从而保证了该部门取号操作的串行性。
数据库层面的并发控制
尽管应用层面的ConcurrentHashMap锁机制可以保证同一部门内的取号逻辑串行执行，但它并不能完全替代数据库层面的并发控制。尤其是在getNextTurnForDepartment方法中，从数据库查询并“锁定”下一个可用号码的操作，仍然可能面临并发问题，例如：
- SELECT 和 UPDATE 非原子性： 如果getNextTurnForDepartment仅执行SELECT查询
