diff --git a/CHANGELOG.md b/CHANGELOG.md
index 110d7d51..1b15c546 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,10 +2,18 @@
 
 ## Unreleased
 
-### Breaking changes
+### Features
 
-- refactor: remove `debug-logs` feature (#820) by @lcian
-  - The deprecated `debug-logs` feature of the `sentry` crate, used for the SDK's own internal logging, has been removed.
+Support for [Sentry structured logs](https://docs.sentry.io/product/explore/logs/) has been added to the SDK.
+- To set up logs, enable the `logs` feature of the `sentry` crate and set `enable_logs` to `true` in your client options.
+- Then, use the `logger_trace!`, `logger_debug!`, `logger_info!`, `logger_warn!`, `logger_error!` and `logger_fatal!` macros to capture logs.
+- To filter or update logs before they are sent, you can use the `before_send_log` client option.
+- Please note that breaking changes could occur until the API is finalized. 
+
+- feat(logs): add log protocol types (#821) by @lcian
+- feat(logs): add ability to capture and send logs (#823) by @lcian & @Swatinem
+- feat(logs): add macro-based API (#827) by @lcian & @szokeasaurusrex
+- feat(logs): send logs in batches (#831) by @lcian
 
 ### Behavioral changes
 
@@ -14,15 +22,10 @@
   - This information is used as a fallback when capturing an event with tracing disabled or otherwise no ongoing span, to still allow related events to be linked by a trace.
   - A new API `Scope::iter_trace_propagation_headers` has been provided that will use the fallback tracing information if there is no current `Span` on the `Scope`.
 
-### Features
+### Breaking changes
 
-- feat(logs): add log protocol types (#821) by @lcian
-- feat(logs): add ability to capture and send logs (#823) by @lcian & @Swatinem
-- feat(logs): add macro-based API (#827) by @lcian & @szokeasaurusrex
-  - Support for [Sentry structured logs](https://docs.sentry.io/product/explore/logs/) has been added.
-  - To enable logs, enable the `UNSTABLE_logs` feature of the `sentry` crate and set `enable_logs` to `true` in your client options.
-  - Then, use the `logger_trace!`, `logger_debug!`, `logger_info!`, `logger_warn!`, `logger_error!` and `logger_fatal!` macros to capture logs.
-  - Please note that breaking changes could occur until the API is finalized.
+- refactor: remove `debug-logs` feature (#820) by @lcian
+  - The deprecated `debug-logs` feature of the `sentry` crate, used for the SDK's own internal logging, has been removed.
 
 ## 0.38.1
 
diff --git a/sentry-core/Cargo.toml b/sentry-core/Cargo.toml
index 784571a2..f3a143cb 100644
--- a/sentry-core/Cargo.toml
+++ b/sentry-core/Cargo.toml
@@ -24,7 +24,7 @@ default = []
 client = ["rand"]
 test = ["client", "release-health"]
 release-health = []
-UNSTABLE_logs = []
+logs = []
 
 [dependencies]
 log = { version = "0.4.8", optional = true, features = ["std"] }
diff --git a/sentry-core/src/client.rs b/sentry-core/src/client.rs
index 10fafa97..c9abf6b2 100644
--- a/sentry-core/src/client.rs
+++ b/sentry-core/src/client.rs
@@ -5,14 +5,14 @@ use std::panic::RefUnwindSafe;
 use std::sync::{Arc, RwLock};
 use std::time::Duration;
 
-#[cfg(feature = "UNSTABLE_logs")]
-use crate::protocol::EnvelopeItem;
 #[cfg(feature = "release-health")]
 use crate::protocol::SessionUpdate;
 use rand::random;
 use sentry_types::random_uuid;
 
 use crate::constants::SDK_INFO;
+#[cfg(feature = "logs")]
+use crate::logs::LogsBatcher;
 use crate::protocol::{ClientSdkInfo, Event};
 #[cfg(feature = "release-health")]
 use crate::session::SessionFlusher;
@@ -20,7 +20,7 @@ use crate::types::{Dsn, Uuid};
 #[cfg(feature = "release-health")]
 use crate::SessionMode;
 use crate::{ClientOptions, Envelope, Hub, Integration, Scope, Transport};
-#[cfg(feature = "UNSTABLE_logs")]
+#[cfg(feature = "logs")]
 use sentry_types::protocol::v7::{Log, LogAttribute};
 
 impl<T: Into<ClientOptions>> From<T> for Client {
@@ -53,6 +53,8 @@ pub struct Client {
     transport: TransportArc,
     #[cfg(feature = "release-health")]
     session_flusher: RwLock<Option<SessionFlusher>>,
+    #[cfg(feature = "logs")]
+    logs_batcher: RwLock<Option<LogsBatcher>>,
     integrations: Vec<(TypeId, Arc<dyn Integration>)>,
     pub(crate) sdk_info: ClientSdkInfo,
 }
@@ -76,11 +78,20 @@ impl Clone for Client {
             self.options.session_mode,
         )));
 
+        #[cfg(feature = "logs")]
+        let logs_batcher = RwLock::new(if self.options.enable_logs {
+            Some(LogsBatcher::new(transport.clone()))
+        } else {
+            None
+        });
+
         Client {
             options: self.options.clone(),
             transport,
             #[cfg(feature = "release-health")]
             session_flusher,
+            #[cfg(feature = "logs")]
+            logs_batcher,
             integrations: self.integrations.clone(),
             sdk_info: self.sdk_info.clone(),
         }
@@ -150,11 +161,20 @@ impl Client {
             options.session_mode,
         )));
 
+        #[cfg(feature = "logs")]
+        let logs_batcher = RwLock::new(if options.enable_logs {
+            Some(LogsBatcher::new(transport.clone()))
+        } else {
+            None
+        });
+
         Client {
             options,
             transport,
             #[cfg(feature = "release-health")]
             session_flusher,
+            #[cfg(feature = "logs")]
+            logs_batcher,
             integrations,
             sdk_info,
         }
@@ -329,6 +349,10 @@ impl Client {
         if let Some(ref flusher) = *self.session_flusher.read().unwrap() {
             flusher.flush();
         }
+        #[cfg(feature = "logs")]
+        if let Some(ref batcher) = *self.logs_batcher.read().unwrap() {
+            batcher.flush();
+        }
         if let Some(ref transport) = *self.transport.read().unwrap() {
             transport.flush(timeout.unwrap_or(self.options.shutdown_timeout))
         } else {
@@ -346,6 +370,8 @@ impl Client {
     pub fn close(&self, timeout: Option<Duration>) -> bool {
         #[cfg(feature = "release-health")]
         drop(self.session_flusher.write().unwrap().take());
+        #[cfg(feature = "logs")]
+        drop(self.logs_batcher.write().unwrap().take());
         let transport_opt = self.transport.write().unwrap().take();
         if let Some(transport) = transport_opt {
             sentry_debug!("client close; request transport to shut down");
@@ -369,24 +395,21 @@ impl Client {
     }
 
     /// Captures a log and sends it to Sentry.
-    #[cfg(feature = "UNSTABLE_logs")]
+    #[cfg(feature = "logs")]
     pub fn capture_log(&self, log: Log, scope: &Scope) {
         if !self.options().enable_logs {
             return;
         }
-        if let Some(ref transport) = *self.transport.read().unwrap() {
-            if let Some(log) = self.prepare_log(log, scope) {
-                let mut envelope = Envelope::new();
-                let logs: EnvelopeItem = vec![log].into();
-                envelope.add_item(logs);
-                transport.send_envelope(envelope);
+        if let Some(log) = self.prepare_log(log, scope) {
+            if let Some(ref batcher) = *self.logs_batcher.read().unwrap() {
+                batcher.enqueue(log);
             }
         }
     }
 
     /// Prepares a log to be sent, setting the `trace_id` and other default attributes, and
     /// processing it through `before_send_log`.
-    #[cfg(feature = "UNSTABLE_logs")]
+    #[cfg(feature = "logs")]
     fn prepare_log(&self, mut log: Log, scope: &Scope) -> Option<Log> {
         scope.apply_to_log(&mut log, self.options.send_default_pii);
 
@@ -399,7 +422,7 @@ impl Client {
         Some(log)
     }
 
-    #[cfg(feature = "UNSTABLE_logs")]
+    #[cfg(feature = "logs")]
     fn set_log_default_attributes(&self, log: &mut Log) {
         if !log.attributes.contains_key("sentry.environment") {
             if let Some(environment) = self.options.environment.as_ref() {
diff --git a/sentry-core/src/clientoptions.rs b/sentry-core/src/clientoptions.rs
index a741cf43..a3407788 100644
--- a/sentry-core/src/clientoptions.rs
+++ b/sentry-core/src/clientoptions.rs
@@ -5,7 +5,7 @@ use std::time::Duration;
 
 use crate::constants::USER_AGENT;
 use crate::performance::TracesSampler;
-#[cfg(feature = "UNSTABLE_logs")]
+#[cfg(feature = "logs")]
 use crate::protocol::Log;
 use crate::protocol::{Breadcrumb, Event};
 use crate::types::Dsn;
@@ -147,7 +147,7 @@ pub struct ClientOptions {
     /// Callback that is executed for each Breadcrumb being added.
     pub before_breadcrumb: Option<BeforeCallback<Breadcrumb>>,
     /// Callback that is executed for each Log being added.
-    #[cfg(feature = "UNSTABLE_logs")]
+    #[cfg(feature = "logs")]
     pub before_send_log: Option<BeforeCallback<Log>>,
     // Transport options
     /// The transport to use.
@@ -171,7 +171,7 @@ pub struct ClientOptions {
     /// server integrations. Needs `send_default_pii` to be enabled to have any effect.
     pub max_request_body_size: MaxRequestBodySize,
     /// Determines whether captured structured logs should be sent to Sentry (defaults to false).
-    #[cfg(feature = "UNSTABLE_logs")]
+    #[cfg(feature = "logs")]
     pub enable_logs: bool,
     // Other options not documented in Unified API
     /// Disable SSL verification.
@@ -232,7 +232,7 @@ impl fmt::Debug for ClientOptions {
         #[derive(Debug)]
         struct BeforeBreadcrumb;
         let before_breadcrumb = self.before_breadcrumb.as_ref().map(|_| BeforeBreadcrumb);
-        #[cfg(feature = "UNSTABLE_logs")]
+        #[cfg(feature = "logs")]
         let before_send_log = {
             #[derive(Debug)]
             struct BeforeSendLog;
@@ -279,7 +279,7 @@ impl fmt::Debug for ClientOptions {
             .field("auto_session_tracking", &self.auto_session_tracking)
             .field("session_mode", &self.session_mode);
 
-        #[cfg(feature = "UNSTABLE_logs")]
+        #[cfg(feature = "logs")]
         debug_struct
             .field("enable_logs", &self.enable_logs)
             .field("before_send_log", &before_send_log);
@@ -325,9 +325,9 @@ impl Default for ClientOptions {
             trim_backtraces: true,
             user_agent: Cow::Borrowed(USER_AGENT),
             max_request_body_size: MaxRequestBodySize::Medium,
-            #[cfg(feature = "UNSTABLE_logs")]
+            #[cfg(feature = "logs")]
             enable_logs: false,
-            #[cfg(feature = "UNSTABLE_logs")]
+            #[cfg(feature = "logs")]
             before_send_log: None,
         }
     }
diff --git a/sentry-core/src/hub.rs b/sentry-core/src/hub.rs
index fe246078..ebd70b0c 100644
--- a/sentry-core/src/hub.rs
+++ b/sentry-core/src/hub.rs
@@ -247,7 +247,7 @@ impl Hub {
     }
 
     /// Captures a structured log.
-    #[cfg(feature = "UNSTABLE_logs")]
+    #[cfg(feature = "logs")]
     pub fn capture_log(&self, log: Log) {
         with_client_impl! {{
             let top = self.inner.with(|stack| stack.top().clone());
diff --git a/sentry-core/src/lib.rs b/sentry-core/src/lib.rs
index d21cc6ba..a39e63ca 100644
--- a/sentry-core/src/lib.rs
+++ b/sentry-core/src/lib.rs
@@ -132,7 +132,7 @@ pub use crate::intodsn::IntoDsn;
 pub use crate::performance::*;
 pub use crate::scope::{Scope, ScopeGuard};
 pub use crate::transport::{Transport, TransportFactory};
-#[cfg(feature = "UNSTABLE_logs")]
+#[cfg(feature = "logs")]
 mod logger; // structured logging macros exported with `#[macro_export]`
 
 // client feature
@@ -140,6 +140,8 @@ mod logger; // structured logging macros exported with `#[macro_export]`
 mod client;
 #[cfg(feature = "client")]
 mod hub_impl;
+#[cfg(all(feature = "client", feature = "logs"))]
+mod logs;
 #[cfg(feature = "client")]
 mod session;
 
diff --git a/sentry-core/src/logs.rs b/sentry-core/src/logs.rs
new file mode 100644
index 00000000..9be3ee33
--- /dev/null
+++ b/sentry-core/src/logs.rs
@@ -0,0 +1,197 @@
+//! Batching for Sentry [structured logs](https://docs.sentry.io/product/explore/logs/).
+
+use std::sync::{Arc, Condvar, Mutex, MutexGuard};
+use std::thread::JoinHandle;
+use std::time::{Duration, Instant};
+
+use crate::client::TransportArc;
+use crate::protocol::EnvelopeItem;
+use crate::Envelope;
+use sentry_types::protocol::v7::Log;
+
+// Flush when there's 100 logs in the buffer
+const MAX_LOG_ITEMS: usize = 100;
+// Or when 5 seconds have passed from the last flush
+const FLUSH_INTERVAL: Duration = Duration::from_secs(5);
+
+#[derive(Debug, Default)]
+struct LogQueue {
+    logs: Vec<Log>,
+}
+
+/// Accumulates logs in the queue and submits them through the transport when one of the flushing
+/// conditions is met.
+pub(crate) struct LogsBatcher {
+    transport: TransportArc,
+    queue: Arc<Mutex<LogQueue>>,
+    shutdown: Arc<(Mutex<bool>, Condvar)>,
+    worker: Option<JoinHandle<()>>,
+}
+
+impl LogsBatcher {
+    /// Creates a new LogsBatcher that will submit envelopes to the given `transport`.
+    pub(crate) fn new(transport: TransportArc) -> Self {
+        let queue = Arc::new(Mutex::new(Default::default()));
+        #[allow(clippy::mutex_atomic)]
+        let shutdown = Arc::new((Mutex::new(false), Condvar::new()));
+
+        let worker_transport = transport.clone();
+        let worker_queue = queue.clone();
+        let worker_shutdown = shutdown.clone();
+        let worker = std::thread::Builder::new()
+            .name("sentry-logs-batcher".into())
+            .spawn(move || {
+                let (lock, cvar) = worker_shutdown.as_ref();
+                let mut shutdown = lock.lock().unwrap();
+                // check this immediately, in case the main thread is already shutting down
+                if *shutdown {
+                    return;
+                }
+                let mut last_flush = Instant::now();
+                loop {
+                    let timeout = FLUSH_INTERVAL
+                        .checked_sub(last_flush.elapsed())
+                        .unwrap_or_else(|| Duration::from_secs(0));
+                    shutdown = cvar.wait_timeout(shutdown, timeout).unwrap().0;
+                    if *shutdown {
+                        return;
+                    }
+                    if last_flush.elapsed() >= FLUSH_INTERVAL {
+                        LogsBatcher::flush_queue_internal(
+                            worker_queue.lock().unwrap(),
+                            &worker_transport,
+                        );
+                        last_flush = Instant::now();
+                    }
+                }
+            })
+            .unwrap();
+
+        Self {
+            transport,
+            queue,
+            shutdown,
+            worker: Some(worker),
+        }
+    }
+
+    /// Enqueues a log for delayed sending.
+    ///
+    /// This will automatically flush the queue if it reaches a size of `BATCH_SIZE`.
+    pub(crate) fn enqueue(&self, log: Log) {
+        let mut queue = self.queue.lock().unwrap();
+        queue.logs.push(log);
+        if queue.logs.len() >= MAX_LOG_ITEMS {
+            LogsBatcher::flush_queue_internal(queue, &self.transport);
+        }
+    }
+
+    /// Flushes the queue to the transport.
+    pub(crate) fn flush(&self) {
+        let queue = self.queue.lock().unwrap();
+        LogsBatcher::flush_queue_internal(queue, &self.transport);
+    }
+
+    /// Flushes the queue to the transport.
+    ///
+    /// This is a static method as it will be called from both the background
+    /// thread and the main thread on drop.
+    fn flush_queue_internal(mut queue_lock: MutexGuard<LogQueue>, transport: &TransportArc) {
+        let logs = std::mem::take(&mut queue_lock.logs);
+        drop(queue_lock);
+
+        if logs.is_empty() {
+            return;
+        }
+
+        sentry_debug!("[LogsBatcher] Flushing {} logs", logs.len());
+
+        if let Some(ref transport) = *transport.read().unwrap() {
+            let mut envelope = Envelope::new();
+            let logs_item: EnvelopeItem = logs.into();
+            envelope.add_item(logs_item);
+            transport.send_envelope(envelope);
+        }
+    }
+}
+
+impl Drop for LogsBatcher {
+    fn drop(&mut self) {
+        let (lock, cvar) = self.shutdown.as_ref();
+        *lock.lock().unwrap() = true;
+        cvar.notify_one();
+
+        if let Some(worker) = self.worker.take() {
+            worker.join().ok();
+        }
+        LogsBatcher::flush_queue_internal(self.queue.lock().unwrap(), &self.transport);
+    }
+}
+
+#[cfg(all(test, feature = "test"))]
+mod tests {
+    use crate::logger_info;
+    use crate::test;
+
+    // Test that logs are sent in batches
+    #[test]
+    fn test_logs_batching() {
+        let envelopes = test::with_captured_envelopes_options(
+            || {
+                for i in 0..150 {
+                    logger_info!("test log {}", i);
+                }
+            },
+            crate::ClientOptions {
+                enable_logs: true,
+                ..Default::default()
+            },
+        );
+
+        assert_eq!(2, envelopes.len());
+
+        let mut total_logs = 0;
+        for envelope in &envelopes {
+            for item in envelope.items() {
+                if let crate::protocol::EnvelopeItem::ItemContainer(
+                    crate::protocol::ItemContainer::Logs(logs),
+                ) = item
+                {
+                    total_logs += logs.len();
+                }
+            }
+        }
+
+        assert_eq!(150, total_logs);
+    }
+
+    // Test that the batcher is flushed on client close
+    #[test]
+    fn test_logs_batcher_flush() {
+        let envelopes = test::with_captured_envelopes_options(
+            || {
+                for i in 0..12 {
+                    logger_info!("test log {}", i);
+                }
+            },
+            crate::ClientOptions {
+                enable_logs: true,
+                ..Default::default()
+            },
+        );
+
+        assert_eq!(1, envelopes.len());
+
+        for envelope in &envelopes {
+            for item in envelope.items() {
+                if let crate::protocol::EnvelopeItem::ItemContainer(
+                    crate::protocol::ItemContainer::Logs(logs),
+                ) = item
+                {
+                    assert_eq!(12, logs.len());
+                    break;
+                }
+            }
+        }
+    }
+}
diff --git a/sentry-core/src/scope/noop.rs b/sentry-core/src/scope/noop.rs
index 578c1f9b..fc62120b 100644
--- a/sentry-core/src/scope/noop.rs
+++ b/sentry-core/src/scope/noop.rs
@@ -1,6 +1,6 @@
 use std::fmt;
 
-#[cfg(feature = "UNSTABLE_logs")]
+#[cfg(feature = "logs")]
 use crate::protocol::Log;
 use crate::protocol::{Context, Event, Level, User, Value};
 use crate::TransactionOrSpan;
@@ -113,7 +113,7 @@ impl Scope {
     }
 
     /// Applies the contained scoped data to fill a log.
-    #[cfg(feature = "UNSTABLE_logs")]
+    #[cfg(feature = "logs")]
     pub fn apply_to_log(&self, log: &mut Log) {
         let _log = log;
         minimal_unreachable!();
diff --git a/sentry-core/src/scope/real.rs b/sentry-core/src/scope/real.rs
index c8147ce0..f4972cf1 100644
--- a/sentry-core/src/scope/real.rs
+++ b/sentry-core/src/scope/real.rs
@@ -9,7 +9,7 @@ use crate::performance::TransactionOrSpan;
 use crate::protocol::{
     Attachment, Breadcrumb, Context, Event, Level, TraceContext, Transaction, User, Value,
 };
-#[cfg(feature = "UNSTABLE_logs")]
+#[cfg(feature = "logs")]
 use crate::protocol::{Log, LogAttribute};
 #[cfg(feature = "release-health")]
 use crate::session::Session;
@@ -350,7 +350,7 @@ impl Scope {
 
     /// Applies the contained scoped data to a log, setting the `trace_id` and certain default
     /// attributes.
-    #[cfg(feature = "UNSTABLE_logs")]
+    #[cfg(feature = "logs")]
     pub fn apply_to_log(&self, log: &mut Log, send_default_pii: bool) {
         if let Some(span) = self.span.as_ref() {
             log.trace_id = Some(span.get_trace_context().trace_id);
diff --git a/sentry-types/src/protocol/envelope.rs b/sentry-types/src/protocol/envelope.rs
index 43a9c35f..190c439b 100644
--- a/sentry-types/src/protocol/envelope.rs
+++ b/sentry-types/src/protocol/envelope.rs
@@ -308,7 +308,7 @@ impl Envelope {
     }
 
     /// Create an [`Iterator`] over all the [`EnvelopeItem`]s.
-    pub fn items(&self) -> EnvelopeItemIter {
+    pub fn items(&self) -> EnvelopeItemIter<'_> {
         let inner = match &self.items {
             Items::EnvelopeItems(items) => items.iter(),
             Items::Raw(_) => [].iter(),
diff --git a/sentry/Cargo.toml b/sentry/Cargo.toml
index 9ab7e315..7737c31a 100644
--- a/sentry/Cargo.toml
+++ b/sentry/Cargo.toml
@@ -48,7 +48,7 @@ opentelemetry = ["sentry-opentelemetry"]
 # other features
 test = ["sentry-core/test"]
 release-health = ["sentry-core/release-health", "sentry-actix?/release-health"]
-UNSTABLE_logs = ["sentry-core/UNSTABLE_logs"]
+logs = ["sentry-core/logs"]
 # transports
 transport = ["reqwest", "native-tls"]
 reqwest = ["dep:reqwest", "httpdate", "tokio"]
diff --git a/sentry/tests/test_basic.rs b/sentry/tests/test_basic.rs
index fb80c90d..8f5f4d0b 100644
--- a/sentry/tests/test_basic.rs
+++ b/sentry/tests/test_basic.rs
@@ -265,7 +265,7 @@ fn test_panic_scope_pop() {
     );
 }
 
-#[cfg(feature = "UNSTABLE_logs")]
+#[cfg(feature = "logs")]
 #[test]
 fn test_basic_capture_log() {
     use std::time::SystemTime;
@@ -314,7 +314,7 @@ fn test_basic_capture_log() {
     }
 }
 
-#[cfg(feature = "UNSTABLE_logs")]
+#[cfg(feature = "logs")]
 #[test]
 fn test_basic_capture_log_macro_message() {
     use sentry_core::logger_info;
@@ -349,7 +349,7 @@ fn test_basic_capture_log_macro_message() {
     }
 }
 
-#[cfg(feature = "UNSTABLE_logs")]
+#[cfg(feature = "logs")]
 #[test]
 fn test_basic_capture_log_macro_message_formatted() {
     use sentry::protocol::LogAttribute;
@@ -416,7 +416,7 @@ fn test_basic_capture_log_macro_message_formatted() {
     }
 }
 
-#[cfg(feature = "UNSTABLE_logs")]
+#[cfg(feature = "logs")]
 #[test]
 fn test_basic_capture_log_macro_message_with_attributes() {
     use sentry::protocol::LogAttribute;
@@ -475,7 +475,7 @@ fn test_basic_capture_log_macro_message_with_attributes() {
     }
 }
 
-#[cfg(feature = "UNSTABLE_logs")]
+#[cfg(feature = "logs")]
 #[test]
 fn test_basic_capture_log_macro_message_formatted_with_attributes() {
     use sentry::protocol::LogAttribute;