已有 PM2.5 數據 API 接口，需定期拉取數據存入自身數據庫。本文提供從 HTTP 請求、JSON 解析、數據清洗到入庫的完整工程化流程，涵蓋 Java/Android/通用三種技術棧方案。
<dependency>
<groupId>org.apache.httpcomponents.client5</groupId>
<artifactId>httpclient5</artifactId>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
fastjson2
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>@Service
@Slf4j
public class Pm25DataSyncService {
@Value("${pm25.api.url}")
private String apiUrl;
@Value("${pm25.api.key}")
private String apiKey;
@Autowired
private JdbcTemplate jdbcTemplate;
@Scheduled(cron = "0 */10 * * * ?") // 每 10 分鐘
public void syncPm25Data() {
try {
// 1. 構建請求
String url = apiUrl + "?key=" + apiKey + "&city=all";
// 2. 發送請求 (HttpClient 5.x)
try (CloseableHttpClient client = HttpClients.createDefault()) {
HttpGet request = new HttpGet(url);
request.setHeader("Accept", "application/json");
try (ClassicHttpResponse response = client.executeOpen(null, request, null)) {
int status = response.getCode();
if (status != 200) {
log.error("API 請求失敗: HTTP {}", status);
return;
}
String json = EntityUtils.toString(response.getEntity());
// 3. 解析 JSON (FastJSON2)
JSONObject data = JSON.parseObject(json);
JSONArray stations = data.getJSONArray("stations"); // 視具體 API 結構
// 4. 批量準備入庫數據
List batchArgs = new ArrayList<>();
for (Object obj : stations) {
JSONObject station = (JSONObject) obj;
// 數據清洗與校驗
String city = station.getString("city");
Double pm25 = station.getDouble("pm25");
Long timestamp = station.getLong("time");
if (city == null || pm25 == null || pm25 < 0 || pm25 > 1000) continue;
batchArgs.add(new Object[]{
city, pm25, new Timestamp(timestamp * 1000),
new Date(), station.getString("station_id")
});
}
// 5. 批量入庫 (ON DUPLICATE KEY UPDATE)
String sql = "INSERT INTO pm25_data (city, pm25_value, measure_time, created_at, station_id) " +
"VALUES (?, ?, ?, ?, ?) " +
"ON DUPLICATE KEY UPDATE pm25_value=VALUES(pm25_value), measure_time=VALUES(measure_time)";
int[] results = jdbcTemplate.batchUpdate(sql, batchArgs);
log.info("PM2.5 數據同步完成: 成功 {} 條", Arrays.stream(results).sum());
}
}
} catch (Exception e) {
log.error("PM2.5 同步異常", e);
// 發送告警通知
}
}
} // 1. 定時任務 (WorkManager)
val workRequest = PeriodicWorkRequestBuilder(15, TimeUnit.MINUTES).build()
WorkManager.getInstance(context).enqueueUniquePeriodicWork(
"pm25_sync", ExistingPeriodicWorkPolicy.KEEP, workRequest)
// 2. Worker 實現
class Pm25SyncWorker @AssistedInject constructor(
@Assisted context: Context, @Assisted params: WorkerParameters,
private val apiService: Pm25ApiService, // Retrofit
private val repository: Pm25Repository // Room
) : CoroutineWorker(context, params) {
override suspend fun doWork(): Result {
return try {
val response = apiService.getAllCities().await()
if (!response.isSuccessful) return Result.retry()
val stations = response.body()?.stations ?: emptyList()
val entities = stations.mapNotNull { dto ->
dto.pm25?.let {
Pm25Entity(
city = dto.city,
pm25 = it,
time = dto.time * 1000L,
stationId = dto.stationId
)
}
}
repository.insertAll(entities) // Room @Insert(onConflict = REPLACE)
Result.success()
} catch (e: Exception) {
Result.retry()
}
}
} import requests
import json
from sqlalchemy import create_engine, text
from datetime import datetime
API_URL = "https://api.pm25.example.com/data"
DB_URL = "mysql+pymysql://user:pass@host/db"
engine = create_engine(DB_URL)
def sync():
try:
resp = requests.get(API_URL, params={"key": "YOUR_KEY"}, timeout=10)
resp.raise_for_status()
data = resp.json()
stations = data.get("stations", [])
rows = []
for s in stations:
pm25 = s.get("pm25")
if pm25 is None or not (0 <= pm25 <= 1000): continue
rows.append({
"city": s["city"],
"pm25_value": pm25,
"measure_time": datetime.fromtimestamp(s["time"]),
"station_id": s.get("station_id")
})
if rows:
with engine.begin() as conn:
# 批量 UPSERT
for row in rows:
conn.execute(text("""
INSERT INTO pm25_data (city, pm25_value, measure_time, station_id)
VALUES (:city, :pm25_value, :measure_time, :station_id)
ON DUPLICATE KEY UPDATE pm25_value=:pm25_value, measure_time=:measure_time
"""), row)
print(f"同步 {len(rows)} 條記錄")
except Exception as e:
print(f"同步失敗: {e}")
if __name__ == "__main__":
sync()CREATE TABLE pm25_data (
id BIGINT AUTO_INCREMENT PRIMARY KEY,
city VARCHAR(50) NOT NULL,
station_id VARCHAR(50),
pm25_value DECIMAL(6,2) NOT NULL,
measure_time DATETIME NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
UNIQUE KEY uk_city_station_time (city, station_id, measure_time),
INDEX idx_city_time (city, measure_time)
) ENGINE=InnoDB PARTITION BY RANGE (TO_DAYS(measure_time)) (
PARTITION p202401 VALUES LESS THAN (TO_DAYS('2024-02-01')),
...
);核心思想：標準化流程 + 批量操作 + 冒等設計 + 完整觀測。根據團隊技術棧選擇 Java/Android/Python 方案，原理相通。
