Организация масштабируемого хранилища с Apache Cassandra

// Пример настройки уровня согласованности в Java-драйвере
Statement statement = new SimpleStatement("SELECT * FROM sensor_data WHERE sensor_id = ?", sensorId);
statement.setConsistencyLevel(ConsistencyLevel.QUORUM);
ResultSet results = session.execute(statement);

JVM_OPTS="$JVM_OPTS -XX:+AlwaysPreTouch"
JVM_OPTS="$JVM_OPTS -XX:+PerfDisableSharedMem"
JVM_OPTS="$JVM_OPTS -XX:+UseNUMA"
JVM_OPTS="$JVM_OPTS -XX:+ExplicitGCInvokesConcurrent"

// Подключение к кластеру
Cluster cluster = Cluster.builder()
        .addContactPoints("node1", "node2", "node3")
        .withCredentials("username", "password")
        .withLoadBalancingPolicy(
            new TokenAwarePolicy(new DCAwareRoundRobinPolicy.Builder().build()))
        .build();
 
// Создание сессии
Session session = cluster.connect("weather_keyspace");
 
// Выполнение запроса
ResultSet results = session.execute("SELECT * FROM temperature_readings " + 
                              "WHERE station_id = ? AND date = ?", 
                              stationId, today);
 
// Обработка результатов
for (Row row : results) {
    float temperature = row.getFloat("temperature");
    Timestamp time = row.getTimestamp("time");
    // Дальнейшая обработка...
}

// Асинхронное выполнение запроса
ResultSetFuture future = session.executeAsync("SELECT * FROM sensor_data " + 
                                      "WHERE sensor_id = ? AND day = ?", 
                                      sensorId, date);
 
// Добавление колбэка для обработки результатов
Futures.addCallback(future, new FutureCallback<ResultSet>() {
    @Override
    public void onSuccess(ResultSet results) {
        for (Row row : results) {
            // Обработка результатов...
        }
    }
 
    @Override
    public void onFailure(Throwable t) {
        // Обработка ошибок...
    }
}, MoreExecutors.directExecutor());
 
// Код продолжает выполняться, не блокируясь

@Service
public class WeatherDataService {
 
    private final CassandraTemplate cassandraTemplate;
    
    // Конструктор с инъекцией зависимостей...
    
    @Transactional
    public void saveWeatherReading(WeatherReading reading) {
        // Сохраняем в текущие показания
        CurrentWeather current = new CurrentWeather();
        current.setStationId(reading.getStationId());
        current.setLastUpdated(new Date());
        current.setTemperature(reading.getTemperature());
        // ... заполнение остальных полей
        cassandraTemplate.insert(current);
        
        // Сохраняем в историю
        WeatherHistory history = convertToHistory(reading);
        cassandraTemplate.insert(history);
        
        // Сохраняем в таблицу экстремумов
        TemperatureExtreme extreme = convertToExtreme(reading);
        cassandraTemplate.insert(extreme);
    }
    
    // Вспомогательные методы для преобразования моделей...
    
    public List<WeatherHistory> getStationHistory(UUID stationId, Date day, int limit) {
        return cassandraTemplate.select(
            Query.query(Criteria.where("station_id").is(stationId)
                .and("day").is(day))
                .limit(limit),
            WeatherHistory.class);
    }
    
    public List<TemperatureExtreme> getHottestStations(Date day, int limit) {
        return cassandraTemplate.select(
            Query.query(Criteria.where("day").is(day))
                .limit(limit),
            TemperatureExtreme.class);
    }
}

PreparedStatement statement = session.prepare(
    "UPDATE users SET email = ? WHERE username = ? IF email = ?");
 
ResultSet rs = session.execute(
    statement.bind("new@example.com", "john_doe", "old@example.com"));
 
if (rs.one().getBool("[applied]")) {
    System.out.println("Email успешно обновлён");
} else {
    System.out.println("Не удалось обновить email, возможно, он был изменён другим процессом");
}

RetryPolicy retryPolicy = new LoggingRetryPolicy(DefaultRetryPolicy.INSTANCE);
 
Cluster cluster = Cluster.builder()
        .addContactPoints("node1", "node2", "node3")
        .withRetryPolicy(retryPolicy)
        .build();
        
// Настройка повторов на уровне запроса
Statement statement = new SimpleStatement("SELECT * FROM sensor_data WHERE sensor_id = ?", sensorId)
        .setConsistencyLevel(ConsistencyLevel.QUORUM)
        .setRetryPolicy(retryPolicy);
        
ResultSet results = session.execute(statement);

@Service
public class CachedWeatherService {
 
    private final WeatherDataService dataService;
    private final Cache<String, CurrentWeather> currentWeatherCache;
    
    public CachedWeatherService(WeatherDataService dataService) {
        this.dataService = dataService;
        this.currentWeatherCache = Caffeine.newBuilder()
            .expireAfterWrite(5, TimeUnit.MINUTES)
            .maximumSize(10_000)
            .build();
    }
    
    public CurrentWeather getCurrentWeather(UUID stationId) {
        String cacheKey = "current:" + stationId.toString();
        return currentWeatherCache.get(cacheKey, key -> {
            // Этот код выполнится только при промахе кэша
            return dataService.getCurrentWeather(stationId);
        });
    }
    
    // Другие методы сервиса...
}

@Repository
public class ReactiveWeatherRepository {
 
    private final ReactiveCassandraTemplate template;
    
    // Конструктор...
    
    public Flux<WeatherHistory> findRecentReadings(UUID stationId, Date day, int limit) {
        return template.select(
            Query.query(Criteria.where("station_id").is(stationId)
                .and("day").is(day))
                .limit(limit),
            WeatherHistory.class);
    }
    
    public Mono<CurrentWeather> save(CurrentWeather weather) {
        return template.insert(weather);
    }
}
 
@RestController
@RequestMapping("/api/weather")
public class WeatherController {
 
    private final ReactiveWeatherRepository repository;
    
    // Конструктор...
    
    @GetMapping("/station/{stationId}/today")
    public Flux<WeatherHistory> getTodaysReadings(@PathVariable UUID stationId) {
        return repository.findRecentReadings(stationId, new Date(), 24);
    }
}