es查询代码es_data_router.get(/search, summary职位搜索) async def search_jobs( # ---------- 关键词 ---------- keyword: Optional[str] Query(None, description搜索关键词如 Java、前端), # ---------- 筛选 ---------- work_location: Optional[str] Query(None, description工作地点精确匹配), edu_require: Optional[str] Query(None, description学历要求), exp_require: Optional[str] Query(None, description经验要求), industry_id: Optional[int] Query(None, description行业ID), company_scale: Optional[int] Query(None, description公司规模枚举值), financing_stage: Optional[int] Query(None, description融资阶段枚举值), status: Optional[int] Query(1, description职位状态默认1招聘中), # ---------- 分页 / 排序 ---------- page: int Query(1, ge1, description页码), page_size: int Query(10, ge1, le50, description每页条数), sort: str Query(time, description排序score相关度time发布时间), es_client: AsyncElasticsearch Depends(es_client_depend), ): 职位搜索思路 1. bool.must/should关键词 multi_match有 keyword 才加 2. bool.filter城市、学历、状态等精确条件不参与算分 3. from/size分页 4. sort有词按相关度或按 publish_time # 索引不存在时友好提示 if not await es_client.indices.exists(indexBOSS_JOB_INDEX_NAME): return { code: 0, message: f索引 {BOSS_JOB_INDEX_NAME} 不存在请先创建并同步数据, } must_clauses: list[dict[str, Any]] [] filter_clauses: list[dict[str, Any]] [] # ----- 1关键词打在多个 text 字段职位名称加权 ----- if keyword and keyword.strip(): must_clauses.append( { multi_match: { query: keyword.strip(), fields: [ job_name^3, # 名称命中权重更高 job_desc^2, duty_require, enterprise_name, industry_name, ], type: best_fields, operator: and, # 可按需改成 or召回更宽 } } ) # ----- 2硬筛选全部放 filter ----- if status is not None: filter_clauses.append({term: {status: status}}) if work_location: filter_clauses.append({term: {work_location: work_location}}) if edu_require: filter_clauses.append({term: {edu_require: edu_require}}) if exp_require: filter_clauses.append({term: {exp_require: exp_require}}) if industry_id is not None: filter_clauses.append({term: {industry_id: industry_id}}) if company_scale is not None: filter_clauses.append({term: {enterpriseInfo_company_scale: company_scale}}) if financing_stage is not None: filter_clauses.append({term: {enterpriseInfo_financing_stage: financing_stage}}) # ----- 3组装 bool ----- bool_query: dict[str, Any] {} if must_clauses: bool_query[must] must_clauses if filter_clauses: bool_query[filter] filter_clauses # 既无关键词也无筛选时匹配全部仍建议至少有默认 status filter query: dict[str, Any] {bool: bool_query} if bool_query else {match_all: {}} # ----- 4排序 ----- # 有关键词且 sortscore → 按相关度否则按发布时间倒序 if sort score and keyword: sort_clause: Any [_score, {publish_time: {order: desc, missing: _last}}] else: sort_clause [{publish_time: {order: desc, missing: _last}}] # ----- 5分页ES 用 from / size ----- from_ (page - 1) * page_size # 列表页只取卡片字段减小 _source source_fields [ job_id, job_name, work_location, min_salary, max_salary, salary_times, edu_require, exp_require, job_tags, status, publish_time, enterprise_id, enterprise_name, enterprise_city_name, enterpriseInfo_company_scale, enterpriseInfo_financing_stage, industry_id, industry_name, ] resp await es_client.search( indexBOSS_JOB_INDEX_NAME, queryquery, sortsort_clause, from_from_, sizepage_size, sourcesource_fields, track_total_hitsTrue, # 拿到准确 total ) hits resp.get(hits, {}) total hits.get(total, {}) # ES 7 total 一般是 {value: n, relation: eq} total_count total.get(value, 0) if isinstance(total, dict) else int(total or 0) lists [] for hit in hits.get(hits, []): item hit.get(_source) or {} item[_score] hit.get(_score) # 可选方便调相关度 lists.append(item) return { code: 1, message: success, data: { lists: lists, page_info: { page: page, page_size: page_size, total_count: total_count, total_page: (total_count page_size - 1) // page_size if page_size else 0, }, }, }雪花算法import time import threading class SnowflakeSingleton: 单例模式的雪花算法生成唯一ID工具类 _instance None _lock threading.Lock() # 单例模式的线程锁 # 起始时间戳 (2020-01-01 00:00:00) START_TIMESTAMP 1577808000000 # 各部分位数 SEQUENCE_BITS 12 # 序列号位数 WORKER_ID_BITS 10 # 工作节点ID位数 # 最大取值计算 MAX_WORKER_ID -1 ^ (-1 WORKER_ID_BITS) # 1023 MAX_SEQUENCE -1 ^ (-1 SEQUENCE_BITS) # 4095 # 移位偏移量计算 WORKER_ID_SHIFT SEQUENCE_BITS # 12 TIMESTAMP_LEFT_SHIFT SEQUENCE_BITS WORKER_ID_BITS # 22 def __new__(cls, worker_idNone): 单例模式实现确保只创建一个实例 with cls._lock: # 如果实例不存在则创建 if not cls._instance: # 首次创建时必须提供worker_id if worker_id is None: raise ValueError(首次初始化必须提供worker_id参数) cls._instance super().__new__(cls) # 初始化实例变量 cls._instance.worker_id worker_id cls._instance.last_timestamp -1 # 上次生成ID的时间戳 cls._instance.sequence 0 # 序列号 cls._instance.id_lock threading.Lock() # 生成ID的线程锁 # 如果已经存在实例再次传入的worker_id必须与初始一致 elif worker_id is not None and worker_id ! cls._instance.worker_id: raise ValueError(f雪花算法实例已初始化worker_id必须为{cls._instance.worker_id}) return cls._instance classmethod def get_instance(cls, worker_idNone): 获取单例实例的便捷方法 return cls(worker_id) def _gen_timestamp(self): 生成当前时间戳毫秒 return int(time.time() * 1000) def get_id(self): 生成下一个唯一ID with self.id_lock: # 确保线程安全 timestamp self._gen_timestamp() # 处理系统时钟回退 if timestamp self.last_timestamp: raise RuntimeError( f系统时钟回退拒绝生成ID。时间差: {self.last_timestamp - timestamp}毫秒 ) # 处理同一毫秒内的序列号 if timestamp self.last_timestamp: self.sequence (self.sequence 1) self.MAX_SEQUENCE # 序列号达到最大值等待下一毫秒 if self.sequence 0: timestamp self._til_next_millis(self.last_timestamp) else: # 不同时间戳序列号重置为0 self.sequence 0 # 更新上次生成ID的时间戳 self.last_timestamp timestamp # 组合生成ID snowflake_id ( ((timestamp - self.START_TIMESTAMP) self.TIMESTAMP_LEFT_SHIFT) | (self.worker_id self.WORKER_ID_SHIFT) | self.sequence ) return snowflake_id def _til_next_millis(self, last_timestamp): 阻塞到下一个毫秒直到获得新的时间戳 timestamp self._gen_timestamp() while timestamp last_timestamp: timestamp self._gen_timestamp() return timestamp # 调用方法示例 if __name__ __main__: # 初始化单例首次调用必须提供worker_id snowflake SnowflakeSingleton.get_instance(worker_id1) # 生成10个ID并打印 for _ in range(10): print(f生成的ID: {snowflake.get_id()}) # 在其他地方获取实例无需再次提供worker_id another_instance SnowflakeSingleton.get_instance() print(f\n验证单例: {snowflake is another_instance}) # 应输出True # 多线程测试 def generate_ids(count): ids [] for _ in range(count): ids.append(SnowflakeSingleton.get_instance().get_id()) return ids # 创建多个线程同时生成ID threads [] for i in range(5): t threading.Thread(targetgenerate_ids, args(1000,)) threads.append(t) t.start() for t in threads: t.join() print(\n多线程ID生成测试完成)